Merge pull request #1593 from benjchristensen/doOnSubscribe

doOnSubscribe

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -4804,6 +4804,21 @@ public final void onNext(T args) {
         return lift(new OperatorDoOnEach<T>(observer));
     }
 
+    /**
+     * Modifies the source {@code Observable} so that it invokes the given action when it is subscribed from
+     * its subscribers. Each subscription will result in an invocation of the given action except when the
+     * source {@code Observable} is reference counted, in which case the source {@code Observable} will invoke
+     * the given action for the first subscription.
+     *
+     *
+     * @param unsubscribe The action that gets called when this {@code Observable} is subscribed.
+     *
+     * @return That modified {@code Observable}
+     */
+    public final Observable<T> doOnSubscribe(final Action0 subscribe) {
+        return lift(new OperatorDoOnSubscribe<T>(subscribe));
+    }
+    
     /**
      * Modifies the source Observable so that it invokes an action when it calls {@code onCompleted} or
      * {@code onError}.
@@ -4845,6 +4860,26 @@ public final void onNext(T args) {
 
         return lift(new OperatorDoOnEach<T>(observer));
     }
+    
+    /**
+     * Modifies the source {@code Observable} so that it invokes the given action when it is unsubscribed from
+     * its subscribers. Each un-subscription will result in an invocation of the given action except when the
+     * source {@code Observable} is reference counted, in which case the source {@code Observable} will invoke
+     * the given action for the very last un-subscription.
+     * <p>
+     * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnUnsubscribe.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doOnUnsubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param unsubscribe the action that gets called when this {@code Observable} is unsubscribed
+     * @return the source {@code Observable} modified so as to call this Action when appropriate
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#doonunsubscribe">RxJava wiki: doOnUnsubscribe</a>
+     */
+    public final Observable<T> doOnUnsubscribe(final Action0 unsubscribe) {
+        return lift(new OperatorDoOnUnsubscribe<T>(unsubscribe));
+    }
 
     /**
      * Returns an Observable that emits the single item at a specified index in a sequence of emissions from a
@@ -10358,23 +10393,4 @@ public void call(Subscriber<? super T> observer) {
         }
     }
 
-    /**
-     * Modifies the source {@code Observable} so that it invokes the given action when it is unsubscribed from
-     * its subscribers. Each un-subscription will result in an invocation of the given action except when the
-     * source {@code Observable} is reference counted, in which case the source {@code Observable} will invoke
-     * the given action for the very last un-subscription.
-     * <p>
-     * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnUnsubscribe.png" alt="">
-     * <dl>
-     *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code doOnUnsubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
-     * </dl>
-     *
-     * @param unsubscribe the action that gets called when this {@code Observable} is unsubscribed
-     * @return the source {@code Observable} modified so as to call this Action when appropriate
-     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#doonunsubscribe">RxJava wiki: doOnUnsubscribe</a>
-     */
-    public final Observable<T> doOnUnsubscribe(final Action0 unsubscribe) {
-        return lift(new OperatorDoOnUnsubscribe<T>(unsubscribe));
-    }
 }

rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnSubscribe.java

@@ -0,0 +1,44 @@
+/**
+ * Copyright 2014 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 rx.internal.operators;
+
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Action0;
+
+/**
+ * This operator modifies an {@link rx.Observable} so a given action is invoked when the {@link rx.Observable} is subscribed.
+ * @param <T> The type of the elements in the {@link rx.Observable} that this operator modifies
+ */
+public class OperatorDoOnSubscribe<T> implements Operator<T, T> {
+    private final Action0 subscribe;
+
+    /**
+     * Constructs an instance of the operator with the callback that gets invoked when the modified Observable is subscribed
+     * @param unsubscribe The action that gets invoked when the modified {@link rx.Observable} is subscribed
+     */
+    public OperatorDoOnSubscribe(Action0 subscribe) {
+        this.subscribe = subscribe;
+    }
+
+    @Override
+    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 child;
+    }
+}

rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnUnsubscribe.java

@@ -15,7 +15,6 @@
  */
 package rx.internal.operators;
 
-import rx.Observable;
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.functions.Action0;

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -1126,125 +1126,4 @@ public String call(Integer t1) {
         ts.assertReceivedOnNext(Arrays.asList("1", "2", "3"));
     }
 
-    @Test
-    public void testDoOnUnsubscribe() throws Exception {
-        int subCount = 3;
-        final CountDownLatch upperLatch = new CountDownLatch(subCount);
-        final CountDownLatch lowerLatch = new CountDownLatch(subCount);
-        final CountDownLatch onNextLatch = new CountDownLatch(subCount);
-
-        final AtomicInteger upperCount = new AtomicInteger();
-        final AtomicInteger lowerCount = new AtomicInteger();
-        Observable<Long> longs = Observable
-            // The stream needs to be infinite to ensure the stream does not terminate
-            // before it is unsubscribed
-            .interval(50, TimeUnit.MILLISECONDS)
-            .doOnUnsubscribe(new Action0() {
-                // Test that upper stream will be notified for un-subscription
-                // from a child subscriber
-                @Override
-                public void call() {
-                    upperLatch.countDown();
-                    upperCount.incrementAndGet();
-                }
-            })
-            .doOnNext(new Action1<Long>() {
-                @Override
-                public void call(Long aLong) {
-                    // Ensure there is at least some onNext events before un-subscription happens
-                    onNextLatch.countDown();
-                }
-            })
-            .doOnUnsubscribe(new Action0() {
-                // Test that lower stream will be notified for a direct un-subscription
-                @Override
-                public void call() {
-                    lowerLatch.countDown();
-                    lowerCount.incrementAndGet();
-                }
-            });
-
-        List<Subscription> subscriptions = new ArrayList<Subscription>();
-        List<TestSubscriber> subscribers = new ArrayList<TestSubscriber>();
-
-        for(int i = 0; i < subCount; ++i) {
-            TestSubscriber<Long> subscriber = new TestSubscriber<Long>();
-            subscriptions.add(longs.subscribe(subscriber));
-            subscribers.add(subscriber);
-        }
-
-        onNextLatch.await();
-        for(int i = 0; i < subCount; ++i) {
-            subscriptions.get(i).unsubscribe();
-            // Test that unsubscribe() method is not affected in any way
-            subscribers.get(i).assertUnsubscribed();
-        }
-
-        upperLatch.await();
-        lowerLatch.await();
-        assertEquals(String.format("There should exactly %d un-subscription events for upper stream", subCount), subCount, upperCount.get());
-        assertEquals(String.format("There should exactly %d un-subscription events for lower stream", subCount), subCount, lowerCount.get());
-    }
-
-    @Test
-    public void testDoOnUnSubscribeWorksWithRefCount() throws Exception {
-        int subCount = 3;
-        final CountDownLatch upperLatch = new CountDownLatch(1);
-        final CountDownLatch lowerLatch = new CountDownLatch(1);
-        final CountDownLatch onNextLatch = new CountDownLatch(subCount);
-
-        final AtomicInteger upperCount = new AtomicInteger();
-        final AtomicInteger lowerCount = new AtomicInteger();
-        Observable<Long> longs = Observable
-            // The stream needs to be infinite to ensure the stream does not terminate
-            // before it is unsubscribed
-            .interval(50, TimeUnit.MILLISECONDS)
-            .doOnUnsubscribe(new Action0() {
-                // Test that upper stream will be notified for un-subscription
-                @Override
-                public void call() {
-                    upperLatch.countDown();
-                    upperCount.incrementAndGet();
-                }
-            })
-            .doOnNext(new Action1<Long>() {
-                @Override
-                public void call(Long aLong) {
-                    // Ensure there is at least some onNext events before un-subscription happens
-                    onNextLatch.countDown();
-                }
-            })
-            .doOnUnsubscribe(new Action0() {
-                // Test that lower stream will be notified for un-subscription
-                @Override
-                public void call() {
-                    lowerLatch.countDown();
-                    lowerCount.incrementAndGet();
-                }
-            })
-            .publish()
-            .refCount();
-
-        List<Subscription> subscriptions = new ArrayList<Subscription>();
-        List<TestSubscriber> subscribers = new ArrayList<TestSubscriber>();
-
-        for(int i = 0; i < subCount; ++i) {
-            TestSubscriber<Long> subscriber = new TestSubscriber<Long>();
-            subscriptions.add(longs.subscribe(subscriber));
-            subscribers.add(subscriber);
-        }
-
-        onNextLatch.await();
-        for(int i = 0; i < subCount; ++i) {
-            subscriptions.get(i).unsubscribe();
-            // Test that unsubscribe() method is not affected in any way
-            subscribers.get(i).assertUnsubscribed();
-        }
-
-        upperLatch.await();
-        lowerLatch.await();
-        assertEquals("There should exactly 1 un-subscription events for upper stream", 1, upperCount.get());
-        assertEquals("There should exactly 1 un-subscription events for lower stream", 1, lowerCount.get());
-
-    }
 }

rxjava-core/src/test/java/rx/internal/operators/OperatorDoOnSubscribeTest.java

@@ -0,0 +1,118 @@
+/**
+ * Copyright 2014 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 rx.internal.operators;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
+import rx.functions.Action0;
+
+public class OperatorDoOnSubscribeTest {
+
+    @Test
+    public void testDoOnSubscribe() throws Exception {
+        final AtomicInteger count = new AtomicInteger();
+        Observable<Integer> o = Observable.just(1).doOnSubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                count.incrementAndGet();
+            }
+
+        });
+
+        o.subscribe();
+        o.subscribe();
+        o.subscribe();
+        assertEquals(3, count.get());
+    }
+
+    @Test
+    public void testDoOnSubscribe2() throws Exception {
+        final AtomicInteger count = new AtomicInteger();
+        Observable<Integer> o = Observable.just(1).doOnSubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                count.incrementAndGet();
+            }
+
+        }).take(1).doOnSubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                count.incrementAndGet();
+            }
+
+        });
+
+        o.subscribe();
+        assertEquals(2, count.get());
+    }
+
+    @Test
+    public void testDoOnUnSubscribeWorksWithRefCount() throws Exception {
+        final AtomicInteger onSubscribed = new AtomicInteger();
+        final AtomicInteger countBefore = new AtomicInteger();
+        final AtomicInteger countAfter = new AtomicInteger();
+        final AtomicReference<Subscriber<? super Integer>> sref = new AtomicReference<Subscriber<? super Integer>>();
+        Observable<Integer> o = Observable.create(new OnSubscribe<Integer>() {
+
+            @Override
+            public void call(Subscriber<? super Integer> s) {
+                onSubscribed.incrementAndGet();
+                sref.set(s);
+            }
+
+        }).doOnSubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                countBefore.incrementAndGet();
+            }
+
+        }).publish().refCount().doOnSubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                countAfter.incrementAndGet();
+            }
+
+        });
+
+        o.subscribe();
+        o.subscribe();
+        o.subscribe();
+        assertEquals(1, countBefore.get());
+        assertEquals(1, onSubscribed.get());
+        assertEquals(3, countAfter.get());
+        sref.get().onCompleted();
+        o.subscribe();
+        o.subscribe();
+        o.subscribe();
+        assertEquals(2, countBefore.get());
+        assertEquals(2, onSubscribed.get());
+        assertEquals(6, countAfter.get());
+    }
+
+}