Merge pull request #3001 from davidmoten/use-subscribers-from

use Subscribers.from()

Author: akarnokd

src/main/java/rx/internal/operators/OperatorDelayWithSelector.java

@@ -20,6 +20,7 @@
 import rx.Subscriber;
 import rx.functions.Func1;
 import rx.observers.SerializedSubscriber;
+import rx.observers.Subscribers;
 import rx.subjects.PublishSubject;
 
 /**
@@ -44,24 +45,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> _child) {
         final SerializedSubscriber<T> child = new SerializedSubscriber<T>(_child);
         final PublishSubject<Observable<T>> delayedEmissions = PublishSubject.create();
 
-        _child.add(Observable.merge(delayedEmissions).unsafeSubscribe(new Subscriber<T>() {
-
-            @Override
-            public void onCompleted() {
-                child.onCompleted();
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                child.onError(e);
-            }
-
-            @Override
-            public void onNext(T t) {
-                child.onNext(t);
-            }
-
-        }));
+        _child.add(Observable.merge(delayedEmissions).unsafeSubscribe(Subscribers.from(child)));
 
         return new Subscriber<T>(_child) {
 

src/main/java/rx/internal/operators/OperatorMulticast.java

@@ -26,6 +26,7 @@
 import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.observables.ConnectableObservable;
+import rx.observers.Subscribers;
 import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
 
@@ -90,22 +91,7 @@ public void connect(Action1<? super Subscription> connection) {
                 final Subject<? super T, ? extends R> subject = subjectFactory.call();
                 // create new Subscriber that will pass-thru to the subject we just created
                 // we do this since it is also a Subscription whereas the Subject is not
-                subscription = new Subscriber<T>() {
-                    @Override
-                    public void onCompleted() {
-                        subject.onCompleted();
-                    }
-
-                    @Override
-                    public void onError(Throwable e) {
-                        subject.onError(e);
-                    }
-
-                    @Override
-                    public void onNext(T args) {
-                        subject.onNext(args);
-                    }
-                };
+                subscription = Subscribers.from(subject);
                 final AtomicReference<Subscription> gs = new AtomicReference<Subscription>();
                 gs.set(Subscriptions.create(new Action0() {
                     @Override