2.x: factor out consumer types, add XConsumable, update ops

Author: akarnokd

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

@@ -0,0 +1,27 @@
+/**
+ * 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;
+
+/**
+ * Represents a basic {@link Completable} source base interface,
+ * consumable via an {@link CompletableSubscriber}.
+ * <p>
+ * This class also serves the base type for custom operators wrapped into
+ * Completable via {@link Completable#create(CompletableConsumable)}.
+ * 
+ * @since 2.0
+ */
+public interface CompletableConsumable {
+    
+    void subscribe(CompletableSubscriber cs);
+}

src/main/java/io/reactivex/CompletableConsumable.java

@@ -0,0 +1,26 @@
+package io.reactivex;
+
+import io.reactivex.disposables.Disposable;
+
+/**
+ * Represents the subscription API callbacks when subscribing to a Completable instance.
+ */
+public interface CompletableSubscriber {
+    /**
+     * Called once the deferred computation completes normally.
+     */
+    void onComplete();
+    
+    /**
+     * Called once if the deferred computation 'throws' an exception.
+     * @param e the exception, not null.
+     */
+    void onError(Throwable e);
+    
+    /**
+     * Called once by the Completable to set a Disposable on this instance which
+     * then can be used to cancel the subscription at any time.
+     * @param d the Disposable instance to call dispose on for cancellation, not null
+     */
+    void onSubscribe(Disposable d);
+}

src/main/java/io/reactivex/CompletableSubscriber.java

@@ -19,7 +19,6 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Single.*;
 import io.reactivex.annotations.*;
 import io.reactivex.disposables.*;
 import io.reactivex.functions.*;
@@ -38,12 +37,8 @@
  * Observable for delivering a sequence of values without backpressure. 
  * @param <T>
  */
-public abstract class Observable<T> {
+public abstract class Observable<T> implements ObservableConsumable<T> {
 
-    public interface NbpOnSubscribe<T> extends Consumer<Observer<? super T>> {
-        
-    }
-    
     public interface NbpOperator<Downstream, Upstream> extends Function<Observer<? super Downstream>, Observer<? super Upstream>> {
 
     }
@@ -53,18 +48,18 @@ public interface NbpTransformer<Upstream, Downstream> extends Function<Observabl
     }
 
     /** An empty observable instance as there is no need to instantiate this more than once. */
-    static final Observable<Object> EMPTY = create(new NbpOnSubscribe<Object>() {
+    static final Observable<Object> EMPTY = create(new ObservableConsumable<Object>() {
         @Override
-        public void accept(Observer<? super Object> s) {
+        public void subscribe(Observer<? super Object> s) {
             s.onSubscribe(EmptyDisposable.INSTANCE);
             s.onComplete();
         }
     });
 
     /** A never NbpObservable instance as there is no need to instantiate this more than once. */
-    static final Observable<Object> NEVER = create(new NbpOnSubscribe<Object>() {
+    static final Observable<Object> NEVER = create(new ObservableConsumable<Object>() {
         @Override
-        public void accept(Observer<? super Object> s) {
+        public void subscribe(Observer<? super Object> s) {
             s.onSubscribe(EmptyDisposable.INSTANCE);
         }
     });
@@ -355,7 +350,7 @@ public static <T> Observable<T> concatArray(Observable<? extends T>... sources)
         return fromArray(sources).concatMap((Function)Functions.identity());
     }
 
-    public static <T> Observable<T> create(NbpOnSubscribe<T> onSubscribe) {
+    public static <T> Observable<T> create(ObservableConsumable<T> onSubscribe) {
         Objects.requireNonNull(onSubscribe, "onSubscribe is null");
         // TODO plugin wrapper
         return new ObservableWrapper<T>(onSubscribe);
@@ -449,9 +444,9 @@ public static <T> Observable<T> fromIterable(Iterable<? extends T> source) {
 
     public static <T> Observable<T> fromPublisher(final Publisher<? extends T> publisher) {
         Objects.requireNonNull(publisher, "publisher is null");
-        return create(new NbpOnSubscribe<T>() {
+        return create(new ObservableConsumable<T>() {
             @Override
-            public void accept(final Observer<? super T> s) {
+            public void subscribe(final Observer<? super T> s) {
                 publisher.subscribe(new Subscriber<T>() {
 
                     @Override
@@ -871,9 +866,9 @@ public static Observable<Integer> range(final int start, final int count) {
         if ((long)start + (count - 1) > Integer.MAX_VALUE) {
             throw new IllegalArgumentException("Integer overflow");
         }
-        return create(new NbpOnSubscribe<Integer>() {
+        return create(new ObservableConsumable<Integer>() {
             @Override
-            public void accept(Observer<? super Integer> s) {
+            public void subscribe(Observer<? super Integer> s) {
                 BooleanDisposable d = new BooleanDisposable();
                 s.onSubscribe(d);
 
@@ -1110,9 +1105,9 @@ public final Observable<Boolean> any(Predicate<? super T> predicate) {
 
     @SchedulerSupport(SchedulerKind.NONE)
     public final Observable<T> asObservable() {
-        return create(new NbpOnSubscribe<T>() {
+        return create(new ObservableConsumable<T>() {
             @Override
-            public void accept(Observer<? super T> s) {
+            public void subscribe(Observer<? super T> s) {
                 Observable.this.subscribe(s);
             }
         });
@@ -2628,6 +2623,7 @@ public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super T
         return ls;
     }
 
+    @Override
     public final void subscribe(Observer<? super T> observer) {
         Objects.requireNonNull(observer, "observer is null");
 
@@ -3154,9 +3150,9 @@ public void request(long n) {
 
     @SchedulerSupport(SchedulerKind.NONE)
     public final Single<T> toSingle() {
-        return Single.create(new SingleOnSubscribe<T>() {
+        return Single.create(new SingleConsumable<T>() {
             @Override
-            public void accept(final SingleSubscriber<? super T> s) {
+            public void subscribe(final SingleSubscriber<? super T> s) {
                 Observable.this.subscribe(new Observer<T>() {
                     T last;
                     @Override

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

@@ -0,0 +1,28 @@
+/**
+ * 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;
+
+/**
+ * Represents a basic, non-backpressured {@link Observable} source base interface,
+ * consumable via an {@link Observer}.
+ * <p>
+ * This class also serves the base type for custom operators wrapped into
+ * Observable via {@link Observable#create(ObservableConsumable)}.
+ * 
+ * @param <T> the element type
+ * @since 2.0
+ */
+public interface ObservableConsumable<T> {
+
+    void subscribe(Observer<? super T> observer);
+}