Merge pull request #3065 from davidmoten/subscribers-wrap

add Subscribers.wrap
ReactiveX · Jul 14, 2015 · bd8fb30 · bd8fb30
2 parents 6e3e344 + a0998b9
commit bd8fb30
Show file tree

Hide file tree

Showing 7 changed files with 50 additions and 87 deletions.
diff --git a/src/main/java/rx/internal/operators/OnSubscribeDefer.java b/src/main/java/rx/internal/operators/OnSubscribeDefer.java
@@ -19,6 +19,7 @@
 import rx.Observable.OnSubscribe;
 import rx.Subscriber;
 import rx.functions.Func0;
+import rx.observers.Subscribers;
 
 /**
  * Do not create the Observable until an Observer subscribes; create a fresh Observable on each
@@ -46,20 +47,7 @@ public void call(final Subscriber<? super T> s) {
             s.onError(t);
             return;
         }
-        o.unsafeSubscribe(new Subscriber<T>(s) {
-            @Override
-            public void onNext(T t) {
-                s.onNext(t);
-            }
-            @Override
-            public void onError(Throwable e) {
-                s.onError(e);
-            }
-            @Override
-            public void onCompleted() {
-                s.onCompleted();
-            }
-        });
+        o.unsafeSubscribe(Subscribers.wrap(s));
     }
 
 }
diff --git a/src/main/java/rx/internal/operators/OnSubscribeDelaySubscription.java b/src/main/java/rx/internal/operators/OnSubscribeDelaySubscription.java
@@ -21,6 +21,7 @@
 import rx.Observable.OnSubscribe;
 import rx.Scheduler.Worker;
 import rx.functions.Action0;
+import rx.observers.Subscribers;
 
 /**
  * Delays the subscription to the source by the given amount, running on the given scheduler.
@@ -49,20 +50,7 @@ public void call(final Subscriber<? super T> s) {
             @Override
             public void call() {
                 if (!s.isUnsubscribed()) {
-                    source.unsafeSubscribe(new Subscriber<T>(s) {
-                        @Override
-                        public void onNext(T t) {
-                            s.onNext(t);
-                        }
-                        @Override
-                        public void onError(Throwable e) {
-                            s.onError(e);
-                        }
-                        @Override
-                        public void onCompleted() {
-                            s.onCompleted();
-                        }
-                    });
+                    source.unsafeSubscribe(Subscribers.wrap(s));
                 }
             }
         }, time, unit);

diff --git a/src/main/java/rx/internal/operators/OnSubscribeDelaySubscriptionWithSelector.java b/src/main/java/rx/internal/operators/OnSubscribeDelaySubscriptionWithSelector.java
@@ -18,6 +18,7 @@
 import rx.*;
 import rx.Observable.OnSubscribe;
 import rx.functions.Func0;
+import rx.observers.Subscribers;
 
 /**
  * Delays the subscription until the Observable<U> emits an event.
@@ -42,20 +43,7 @@ public void call(final Subscriber<? super T> child) {
                 @Override
                 public void onCompleted() {
                     // subscribe to actual source
-                    source.unsafeSubscribe(new Subscriber<T>(child) {
-                        @Override
-                        public void onNext(T t) {
-                            child.onNext(t);
-                        }
-                        @Override
-                        public void onError(Throwable e) {
-                            child.onError(e);
-                        }
-                        @Override
-                        public void onCompleted() {
-                            child.onCompleted();
-                        }
-                    });
+                    source.unsafeSubscribe(Subscribers.wrap(child));
                 }
 
                 @Override

diff --git a/src/main/java/rx/internal/operators/OnSubscribeUsing.java b/src/main/java/rx/internal/operators/OnSubscribeUsing.java
@@ -22,6 +22,7 @@
 import rx.Observable.OnSubscribe;
 import rx.exceptions.CompositeException;
 import rx.functions.*;
+import rx.observers.Subscribers;
 
 /**
  * Constructs an observable sequence that depends on a resource object.
@@ -68,20 +69,7 @@ public void call(final Subscriber<? super T> subscriber) {
                 observable = source;
             try {
                 // start
-                observable.unsafeSubscribe(new Subscriber<T>(subscriber) {
-                    @Override
-                    public void onNext(T t) {
-                        subscriber.onNext(t);
-                    }
-                    @Override
-                    public void onError(Throwable e) {
-                        subscriber.onError(e);
-                    }
-                    @Override
-                    public void onCompleted() {
-                        subscriber.onCompleted();
-                    }
-                });
+                observable.unsafeSubscribe(Subscribers.wrap(subscriber));
             } catch (Throwable e) {
                 Throwable disposeError = disposeEagerlyIfRequested(disposeOnceOnly);
                 if (disposeError != null)

diff --git a/src/main/java/rx/internal/operators/OperatorDoOnSubscribe.java b/src/main/java/rx/internal/operators/OperatorDoOnSubscribe.java
@@ -18,6 +18,7 @@
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.functions.Action0;
+import rx.observers.Subscribers;
 
 /**
  * This operator modifies an {@link rx.Observable} so a given action is invoked when the {@link rx.Observable} is subscribed.
@@ -39,19 +40,6 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
         subscribe.call();
         // Pass through since this operator is for notification only, there is
         // no change to the stream whatsoever.
-        return new Subscriber<T>(child) {
-            @Override
-            public void onNext(T t) {
-                child.onNext(t);
-            }
-            @Override
-            public void onError(Throwable e) {
-                child.onError(e);
-            }
-            @Override
-            public void onCompleted() {
-                child.onCompleted();
-            }
-        };
+        return Subscribers.wrap(child);
     }
 }
diff --git a/src/main/java/rx/internal/operators/OperatorDoOnUnsubscribe.java b/src/main/java/rx/internal/operators/OperatorDoOnUnsubscribe.java
@@ -18,6 +18,7 @@
 import rx.Observable.Operator;
 import rx.*;
 import rx.functions.Action0;
+import rx.observers.Subscribers;
 import rx.subscriptions.Subscriptions;
 
 /**
@@ -41,22 +42,6 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
 
         // Pass through since this operator is for notification only, there is
         // no change to the stream whatsoever.
-        return new Subscriber<T>(child) {
-            @Override
-            public void onStart() {
-            }
-            @Override
-            public void onNext(T t) {
-                child.onNext(t);
-            }
-            @Override
-            public void onError(Throwable e) {
-                child.onError(e);
-            }
-            @Override
-            public void onCompleted() {
-                child.onCompleted();
-            }
-        };
+        return Subscribers.wrap(child);
     }
 }
diff --git a/src/main/java/rx/observers/Subscribers.java b/src/main/java/rx/observers/Subscribers.java
@@ -17,6 +17,7 @@
 
 import rx.Observer;
 import rx.Subscriber;
+import rx.annotations.Experimental;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action0;
 import rx.functions.Action1;
@@ -198,4 +199,41 @@ public final void onNext(T args) {
         };
     }
 
+    /**
+     * Returns a new {@link Subscriber} that passes all events to
+     * <code>subscriber</code>, has backpressure controlled by
+     * <code>subscriber</code> and uses the subscription list of
+     * <code>subscriber</code> when {@link Subscriber#add(rx.Subscription)} is
+     * called.
+     * 
+     * @param subscriber
+     *            the Subscriber to wrap.
+     * 
+     * @return a new Subscriber that passes all events to
+     *         <code>subscriber</code>, has backpressure controlled by
+     *         <code>subscriber</code> and uses <code>subscriber</code> to
+     *         manage unsubscription.
+     * 
+     */
+    @Experimental
+    public static <T> Subscriber<T> wrap(final Subscriber<? super T> subscriber) {
+        return new Subscriber<T>(subscriber) {
+
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                subscriber.onError(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+                subscriber.onNext(t);
+            }
+
+        };
+    }
 }