2.x: Rename Single Base Interface Types for consistency (#4297)

Author: vanniktech

src/main/java/io/reactivex/Completable.java

@@ -310,7 +310,7 @@ public static Completable fromRunnable(final Runnable run) {
      * @throws NullPointerException if single is null
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public static <T> Completable fromSingle(final SingleConsumable<T> single) {
+    public static <T> Completable fromSingle(final SingleSource<T> single) {
         Objects.requireNonNull(single, "single is null");
         return new CompletableFromSingle<T>(single);
     }

src/main/java/io/reactivex/Observable.java

@@ -3152,9 +3152,9 @@ public void request(long n) {
 
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> toSingle() {
-        return Single.create(new SingleConsumable<T>() {
+        return Single.create(new SingleSource<T>() {
             @Override
-            public void subscribe(final SingleSubscriber<? super T> s) {
+            public void subscribe(final SingleObserver<? super T> s) {
                 Observable.this.subscribe(new Observer<T>() {
                     T last;
                     @Override

src/main/java/io/reactivex/Single.java

@@ -15,11 +15,11 @@
 
 import io.reactivex.disposables.Disposable;
 
-public interface SingleSubscriber<T> {
+public interface SingleObserver<T> {
 
     void onSubscribe(Disposable d);
 
     void onSuccess(T value);
 
     void onError(Throwable e);
-}
+}

src/main/java/io/reactivex/SingleSubscriber.java src/main/java/io/reactivex/SingleObserver.java

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex;
+
+import io.reactivex.functions.Function;
+
+public interface SingleOperator<Downstream, Upstream> extends Function<SingleObserver<? super Downstream>, SingleObserver<? super Upstream>> {
+
+}

src/main/java/io/reactivex/SingleOperator.java

@@ -14,15 +14,15 @@
 
 /**
  * Represents a basic {@link Single} source base interface,
- * consumable via an {@link SingleSubscriber}.
+ * consumable via an {@link SingleObserver}.
  * <p>
  * This class also serves the base type for custom operators wrapped into
- * Single via {@link Single#create(SingleConsumable)}.
+ * Single via {@link Single#create(SingleSource)}.
  * 
  * @param <T> the element type
  * @since 2.0
  */
-public interface SingleConsumable<T> {
+public interface SingleSource<T> {
 
-    void subscribe(SingleSubscriber<? super T> s);
+    void subscribe(SingleObserver<? super T> s);
 }

src/main/java/io/reactivex/SingleConsumable.java src/main/java/io/reactivex/SingleSource.java

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex;
+
+import io.reactivex.functions.Function;
+
+public interface SingleTransformer<Upstream, Downstream> extends Function<Single<Upstream>, SingleSource<Downstream>> {
+
+}

src/main/java/io/reactivex/SingleTransformer.java

@@ -50,7 +50,7 @@ public static void error(Throwable e, CompletableSubscriber s) {
         s.onError(e);
     }
 
-    public static void error(Throwable e, SingleSubscriber<?> s) {
+    public static void error(Throwable e, SingleObserver<?> s) {
         s.onSubscribe(INSTANCE);
         s.onError(e);
     }

src/main/java/io/reactivex/internal/disposables/EmptyDisposable.java

@@ -18,15 +18,15 @@
 
 public final class CompletableFromSingle<T> extends Completable {
 
-    final SingleConsumable<T> single;
+    final SingleSource<T> single;
 
-    public CompletableFromSingle(SingleConsumable<T> single) {
+    public CompletableFromSingle(SingleSource<T> single) {
         this.single = single;
     }
 
     @Override
     protected void subscribeActual(final CompletableSubscriber s) {
-        single.subscribe(new SingleSubscriber<T>() {
+        single.subscribe(new SingleObserver<T>() {
 
             @Override
             public void onError(Throwable e) {

src/main/java/io/reactivex/internal/operators/completable/CompletableFromSingle.java

@@ -33,7 +33,7 @@ public CompletableToSingle(CompletableConsumable source,
     }
 
     @Override
-    protected void subscribeActual(final SingleSubscriber<? super T> s) {
+    protected void subscribeActual(final SingleObserver<? super T> s) {
         source.subscribe(new CompletableSubscriber() {
 
             @Override

src/main/java/io/reactivex/internal/operators/completable/CompletableToSingle.java

@@ -21,20 +21,20 @@
 
 public final class SingleAmbArray<T> extends Single<T> {
 
-    final SingleConsumable<? extends T>[] sources;
+    final SingleSource<? extends T>[] sources;
 
-    public SingleAmbArray(SingleConsumable<? extends T>[] sources) {
+    public SingleAmbArray(SingleSource<? extends T>[] sources) {
         this.sources = sources;
     }
 
     @Override
-    protected void subscribeActual(final SingleSubscriber<? super T> s) {
+    protected void subscribeActual(final SingleObserver<? super T> s) {
 
         final AtomicBoolean once = new AtomicBoolean();
         final CompositeDisposable set = new CompositeDisposable();
         s.onSubscribe(set);
 
-        for (SingleConsumable<? extends T> s1 : sources) {
+        for (SingleSource<? extends T> s1 : sources) {
             if (once.get()) {
                 return;
             }
@@ -50,7 +50,7 @@ protected void subscribeActual(final SingleSubscriber<? super T> s) {
                 return;
             }
 
-            s1.subscribe(new SingleSubscriber<T>() {
+            s1.subscribe(new SingleObserver<T>() {
 
                 @Override
                 public void onSubscribe(Disposable d) {

src/main/java/io/reactivex/internal/operators/single/SingleAmbArray.java

@@ -22,18 +22,18 @@
 
 public final class SingleAmbIterable<T> extends Single<T> {
 
-    final Iterable<? extends SingleConsumable<? extends T>> sources;
+    final Iterable<? extends SingleSource<? extends T>> sources;
 
-    public SingleAmbIterable(Iterable<? extends SingleConsumable<? extends T>> sources) {
+    public SingleAmbIterable(Iterable<? extends SingleSource<? extends T>> sources) {
         this.sources = sources;
     }
 
     @Override
-    protected void subscribeActual(final SingleSubscriber<? super T> s) {
+    protected void subscribeActual(final SingleObserver<? super T> s) {
         final CompositeDisposable set = new CompositeDisposable();
         s.onSubscribe(set);
 
-        Iterator<? extends SingleConsumable<? extends T>> iterator;
+        Iterator<? extends SingleSource<? extends T>> iterator;
 
         try {
             iterator = sources.iterator();
@@ -76,7 +76,7 @@ protected void subscribeActual(final SingleSubscriber<? super T> s) {
                 return;
             }
 
-            SingleConsumable<? extends T> s1;
+            SingleSource<? extends T> s1;
 
             try {
                 s1 = iterator.next();
@@ -92,7 +92,7 @@ protected void subscribeActual(final SingleSubscriber<? super T> s) {
                 return;
             }
 
-            s1.subscribe(new SingleSubscriber<T>() {
+            s1.subscribe(new SingleObserver<T>() {
 
                 @Override
                 public void onSubscribe(Disposable d) {

src/main/java/io/reactivex/internal/operators/single/SingleAmbIterable.java

@@ -23,12 +23,12 @@
 public enum SingleAwait {
     ;
 
-    public static <T> T get(SingleConsumable<T> source) {
+    public static <T> T get(SingleSource<T> source) {
         final AtomicReference<T> valueRef = new AtomicReference<T>();
         final AtomicReference<Throwable> errorRef = new AtomicReference<Throwable>();
         final CountDownLatch cdl = new CountDownLatch(1);
 
-        source.subscribe(new SingleSubscriber<T>() {
+        source.subscribe(new SingleObserver<T>() {
             @Override
             public void onError(Throwable e) {
                 errorRef.lazySet(e);

src/main/java/io/reactivex/internal/operators/single/SingleAwait.java

@@ -23,21 +23,21 @@
 
 public final class SingleCache<T> extends Single<T> {
 
-    final SingleConsumable<? extends T> source;
+    final SingleSource<? extends T> source;
 
     final AtomicInteger wip;
     final AtomicReference<Object> notification;
-    final List<SingleSubscriber<? super T>> subscribers;
+    final List<SingleObserver<? super T>> subscribers;
 
-    public SingleCache(SingleConsumable<? extends T> source) {
+    public SingleCache(SingleSource<? extends T> source) {
         this.source = source;
         this.wip = new AtomicInteger();
         this.notification = new AtomicReference<Object>();
-        this.subscribers = new ArrayList<SingleSubscriber<? super T>>();
+        this.subscribers = new ArrayList<SingleObserver<? super T>>();
     }
 
     @Override
-    protected void subscribeActual(SingleSubscriber<? super T> s) {
+    protected void subscribeActual(SingleObserver<? super T> s) {
 
         Object o = notification.get();
         if (o != null) {
@@ -70,7 +70,7 @@ protected void subscribeActual(SingleSubscriber<? super T> s) {
             return;
         }
 
-        source.subscribe(new SingleSubscriber<T>() {
+        source.subscribe(new SingleObserver<T>() {
 
             @Override
             public void onSubscribe(Disposable d) {
@@ -80,25 +80,25 @@ public void onSubscribe(Disposable d) {
             @Override
             public void onSuccess(T value) {
                 notification.set(NotificationLite.next(value));
-                List<SingleSubscriber<? super T>> list;
+                List<SingleObserver<? super T>> list;
                 synchronized (subscribers) {
-                    list = new ArrayList<SingleSubscriber<? super T>>(subscribers);
+                    list = new ArrayList<SingleObserver<? super T>>(subscribers);
                     subscribers.clear();
                 }
-                for (SingleSubscriber<? super T> s1 : list) {
+                for (SingleObserver<? super T> s1 : list) {
                     s1.onSuccess(value);
                 }
             }
 
             @Override
             public void onError(Throwable e) {
                 notification.set(NotificationLite.error(e));
-                List<SingleSubscriber<? super T>> list;
+                List<SingleObserver<? super T>> list;
                 synchronized (subscribers) {
-                    list = new ArrayList<SingleSubscriber<? super T>>(subscribers);
+                    list = new ArrayList<SingleObserver<? super T>>(subscribers);
                     subscribers.clear();
                 }
-                for (SingleSubscriber<? super T> s1 : list) {
+                for (SingleObserver<? super T> s1 : list) {
                     s1.onError(e);
                 }
             }

src/main/java/io/reactivex/internal/operators/single/SingleCache.java

@@ -20,22 +20,22 @@
 
 public final class SingleContains<T> extends Single<Boolean> {
 
-    final SingleConsumable<T> source;
+    final SingleSource<T> source;
 
     final Object value;
 
     final BiPredicate<Object, Object> comparer;
 
-    public SingleContains(SingleConsumable<T> source, Object value, BiPredicate<Object, Object> comparer) {
+    public SingleContains(SingleSource<T> source, Object value, BiPredicate<Object, Object> comparer) {
         this.source = source;
         this.value = value;
         this.comparer = comparer;
     }
 
     @Override
-    protected void subscribeActual(final SingleSubscriber<? super Boolean> s) {
+    protected void subscribeActual(final SingleObserver<? super Boolean> s) {
 
-        source.subscribe(new SingleSubscriber<T>() {
+        source.subscribe(new SingleObserver<T>() {
 
             @Override
             public void onSubscribe(Disposable d) {

src/main/java/io/reactivex/internal/operators/single/SingleContains.java

@@ -20,15 +20,15 @@
 
 public final class SingleDefer<T> extends Single<T> {
 
-    final Callable<? extends SingleConsumable<? extends T>> singleSupplier;
+    final Callable<? extends SingleSource<? extends T>> singleSupplier;
 
-    public SingleDefer(Callable<? extends SingleConsumable<? extends T>> singleSupplier) {
+    public SingleDefer(Callable<? extends SingleSource<? extends T>> singleSupplier) {
         this.singleSupplier = singleSupplier;
     }
 
     @Override
-    protected void subscribeActual(SingleSubscriber<? super T> s) {
-        SingleConsumable<? extends T> next;
+    protected void subscribeActual(SingleObserver<? super T> s) {
+        SingleSource<? extends T> next;
 
         try {
             next = singleSupplier.call();

src/main/java/io/reactivex/internal/operators/single/SingleDefer.java

@@ -21,24 +21,24 @@
 public final class SingleDelay<T> extends Single<T> {
 
 
-    final SingleConsumable<? extends T> source;
+    final SingleSource<? extends T> source;
     final long time;
     final TimeUnit unit;
     final Scheduler scheduler;
 
-    public SingleDelay(SingleConsumable<? extends T> source, long time, TimeUnit unit, Scheduler scheduler) {
+    public SingleDelay(SingleSource<? extends T> source, long time, TimeUnit unit, Scheduler scheduler) {
         this.source = source;
         this.time = time;
         this.unit = unit;
         this.scheduler = scheduler;
     }
 
     @Override
-    protected void subscribeActual(final SingleSubscriber<? super T> s) {
+    protected void subscribeActual(final SingleObserver<? super T> s) {
 
         final SerialDisposable sd = new SerialDisposable();
         s.onSubscribe(sd);
-        source.subscribe(new SingleSubscriber<T>() {
+        source.subscribe(new SingleObserver<T>() {
             @Override
             public void onSubscribe(Disposable d) {
                 sd.replace(d);