Merge pull request #33 from techery/fix-clearReplaysWithSuccess

Fix clearReplays to clear cachedSuccessPipeline. It was missed
techery · Mar 14, 2016 · fe93e0d · fe93e0d
2 parents 2eb2c9e + c3e08e7
commit fe93e0d
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 5 deletions.
diff --git a/janet/src/main/java/io/techery/janet/ActionPipe.java b/janet/src/main/java/io/techery/janet/ActionPipe.java
@@ -97,6 +97,7 @@ public Observable<A> observeSuccessWithReplay() {
      */
     public void clearReplays() {
         createCachedPipeline();
+        createCachedSuccessPipeline();
     }
 
     /**

diff --git a/janet/src/test/java/io/techery/janet/JanetTest.java b/janet/src/test/java/io/techery/janet/JanetTest.java
@@ -16,9 +16,13 @@
 import static java.lang.annotation.ElementType.TYPE;
 import static java.lang.annotation.RetentionPolicy.RUNTIME;
 import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.only;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 public class JanetTest {
@@ -56,8 +60,7 @@ public void sendWithObserve() {
     @Test
     public void sendWithObserveWithReplay() {
         TestSubscriber<ActionState<TestAction>> subscriber = new TestSubscriber<ActionState<TestAction>>();
-        actionPipe.observeWithReplay()
-                .subscribe(subscriber);
+        actionPipe.observeWithReplay().subscribe(subscriber);
         actionPipe.send(new TestAction());
         subscriber.unsubscribe();
         assertSubscriberWithStates(subscriber);
@@ -120,6 +123,7 @@ public void cancelAfterSend() {
         actionPipe.createObservable(action).subscribe(subscriber);
         subscriber.unsubscribe();
         assertSubscriberWithSingleValue(subscriber);
+        verify(service, times(1)).cancel(any(ActionHolder.class));
     }
 
     @Test
@@ -129,9 +133,17 @@ public void clearReplays() {
         TestSubscriber<ActionState<TestAction>> subscriber = new TestSubscriber<ActionState<TestAction>>();
         actionPipe.observeWithReplay().subscribe(subscriber);
         subscriber.unsubscribe();
-        subscriber.assertNoErrors();
-        subscriber.assertNoValues();
-        subscriber.assertUnsubscribed();
+        assertSubscriberWithoutValues(subscriber);
+    }
+
+    @Test
+    public void clearReplaysSuccess() {
+        actionPipe.send(new TestAction());
+        actionPipe.clearReplays();
+        TestSubscriber<TestAction> subscriber = new TestSubscriber<TestAction>();
+        actionPipe.observeSuccessWithReplay().subscribe(subscriber);
+        subscriber.unsubscribe();
+        assertSubscriberWithoutValues(subscriber);
     }
 
     @Test
@@ -161,6 +173,12 @@ private static void assertSubscriberWithSingleValue(TestSubscriber<?> subscriber
         subscriber.assertUnsubscribed();
     }
 
+    private static void assertSubscriberWithoutValues(TestSubscriber<?> subscriber) {
+        subscriber.assertNoErrors();
+        subscriber.assertNoValues();
+        subscriber.assertUnsubscribed();
+    }
+
     private static void assertStatusCount(List<ActionState<TestAction>> values, ActionState.Status status, int count) {
         int i = 0;
         for (ActionState state : values) {