Elide more create/callback wrappers in favor of direct Observables. (#4289)

Author: JakeWharton

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

@@ -451,35 +451,35 @@ 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 ObservableConsumable<T>() {
+        return new Observable<T>() {
             @Override
-            public void subscribe(final Observer<? super T> s) {
+            protected void subscribeActual(final Observer<? super T> o) {
                 publisher.subscribe(new Subscriber<T>() {
 
                     @Override
                     public void onComplete() {
-                        s.onComplete();
+                        o.onComplete();
                     }
 
                     @Override
                     public void onError(Throwable t) {
-                        s.onError(t);
+                        o.onError(t);
                     }
 
                     @Override
                     public void onNext(T t) {
-                        s.onNext(t);
+                        o.onNext(t);
                     }
 
                     @Override
                     public void onSubscribe(Subscription inner) {
-                        s.onSubscribe(Disposables.from(inner));
+                        o.onSubscribe(Disposables.from(inner));
                         inner.request(Long.MAX_VALUE);
                     }
-                    
+
                 });
             }
-        });
+        };
     }
 
     @SchedulerSupport(SchedulerSupport.NONE)
@@ -873,21 +873,21 @@ 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 ObservableConsumable<Integer>() {
+        return new Observable<Integer>() {
             @Override
-            public void subscribe(Observer<? super Integer> s) {
+            protected void subscribeActual(Observer<? super Integer> o) {
                 Disposable d = Disposables.empty();
-                s.onSubscribe(d);
+                o.onSubscribe(d);
 
                 long end = start - 1L + count;
                 for (long i = start; i <= end && !d.isDisposed(); i++) {
-                    s.onNext((int)i);
+                    o.onNext((int)i);
                 }
                 if (!d.isDisposed()) {
-                    s.onComplete();
+                    o.onComplete();
                 }
             }
-        });
+        };
     }
 
     @SchedulerSupport(SchedulerSupport.NONE)
@@ -1116,12 +1116,13 @@ public final Observable<Boolean> any(Predicate<? super T> predicate) {
 
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> asObservable() {
-        return create(new ObservableConsumable<T>() {
+        final Observable<T> outer = this;
+        return new Observable<T>() {
             @Override
-            public void subscribe(Observer<? super T> s) {
-                Observable.this.subscribe(s);
+            protected void subscribeActual(Observer<? super T> o) {
+                outer.subscribe(o);
             }
-        });
+        };
     }
 
     @SchedulerSupport(SchedulerSupport.NONE)

src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java

@@ -119,12 +119,12 @@ public void subscribe(Observer<? super T> child) {
 
     public static <T, R> Observable<R> create(final ObservableConsumable<? extends T> source, 
             final Function<? super Observable<T>, ? extends ObservableConsumable<R>> selector, final int bufferSize) {
-        return create(new ObservableConsumable<R>() {
+        return new Observable<R>() {
             @Override
-            public void subscribe(Observer<? super R> sr) {
-                ConnectableObservable<T> op = create(source, bufferSize);
+            protected void subscribeActual(Observer<? super R> o) {
+                ConnectableObservable<T> op = ObservablePublish.create(source, bufferSize);
 
-                final ObserverResourceWrapper<R> srw = new ObserverResourceWrapper<R>(sr);
+                final ObserverResourceWrapper<R> srw = new ObserverResourceWrapper<R>(o);
 
                 ObservableConsumable<R> target;
 
@@ -145,7 +145,7 @@ public void accept(Disposable r) {
                     }
                 });
             }
-        });
+        };
     }
 
     private ObservablePublish(ObservableConsumable<T> onSubscribe, ObservableConsumable<? extends T> source, 

src/main/java/io/reactivex/internal/operators/observable/ObservableReplay.java

@@ -58,9 +58,9 @@ public Object call() {
     public static <U, R> Observable<R> multicastSelector(
             final Callable<? extends ConnectableObservable<U>> connectableFactory,
             final Function<? super Observable<U>, ? extends ObservableConsumable<R>> selector) {
-        return Observable.create(new ObservableConsumable<R>() {
+        return new Observable<R>() {
             @Override
-            public void subscribe(Observer<? super R> child) {
+            protected void subscribeActual(Observer<? super R> child) {
                 ConnectableObservable<U> co;
                 ObservableConsumable<R> observable;
                 try {
@@ -82,7 +82,7 @@ public void accept(Disposable r) {
                     }
                 });
             }
-        });
+        };
     }
 
     /**

src/main/java/io/reactivex/observables/ConnectableObservable.java

@@ -75,7 +75,7 @@ public void accept(Disposable d) {
      * @see <a href="http://reactivex.io/documentation/operators/refcount.html">ReactiveX documentation: RefCount</a>
      */
     public Observable<T> refCount() {
-        return create(new ObservableRefCount<T>(this));
+        return new ObservableRefCount<T>(this);
     }
 
     /**
@@ -121,6 +121,6 @@ public Observable<T> autoConnect(int numberOfSubscribers, Consumer<? super Dispo
             this.connect(connection);
             return this;
         }
-        return create(new ObservableAutoConnect<T>(this, numberOfSubscribers, connection));
+        return new ObservableAutoConnect<T>(this, numberOfSubscribers, connection);
     }
 }