Call unsubscribe when the terminal events are invoked

Author: akarnokd

README.md

@@ -17,7 +17,7 @@ Library to convert between RxJava 1.x and 2.x reactive types.
 
 ```
 dependencies {
-    compile "com.github.akarnokd:rxjava2-interop:0.11.3"
+    compile "com.github.akarnokd:rxjava2-interop:0.11.4"
 }
 ```
 

gradle.properties

@@ -1 +1 @@
-version=0.11.3
+version=0.11.4

src/main/java/hu/akarnokd/rxjava/interop/ObservableV1ToFlowableV2.java

@@ -71,6 +71,7 @@ public void onError(Throwable e) {
             }
             done = true;
             actual.onError(e);
+            unsubscribe(); // v1 expects an unsubscribe  call when terminated
         }
 
         @Override
@@ -80,6 +81,7 @@ public void onCompleted() {
             }
             done = true;
             actual.onComplete();
+            unsubscribe(); // v1 expects an unsubscribe  call when terminated
         }
 
         void requestMore(long n) {

src/main/java/hu/akarnokd/rxjava/interop/ObservableV1ToObservableV2.java

@@ -70,6 +70,7 @@ public void onError(Throwable e) {
             }
             done = true;
             actual.onError(e);
+            unsubscribe(); // v1 expects an unsubscribe call when terminated
         }
 
         @Override
@@ -79,6 +80,7 @@ public void onCompleted() {
             }
             done = true;
             actual.onComplete();
+            unsubscribe(); // v1 expects an unsubscribe call when terminated
         }
 
         @Override

src/test/java/hu/akarnokd/rxjava/interop/RxJavaInteropTest.java

@@ -1599,4 +1599,53 @@ public void toV1SubscriptionCallsDispose() {
         subscription.unsubscribe();
         verify(disposable).dispose();
     }
+
+    @Test
+    public void v1ObservableIsUnsubscribedOnError() {
+        Action0 onUnsubscribe = mock(Action0.class);
+
+        RxJavaInterop.toV2Observable(rx.Observable.error(new IOException())
+        .doOnUnsubscribe(onUnsubscribe))
+        .test()
+        .assertFailure(IOException.class);
+
+        verify(onUnsubscribe).call();
+    }
+
+
+    @Test
+    public void v1ObservableIsUnsubscribedOnCompletion() {
+        Action0 onUnsubscribe = mock(Action0.class);
+
+        RxJavaInterop.toV2Observable(rx.Observable.just(1)
+        .doOnUnsubscribe(onUnsubscribe))
+        .test()
+        .assertResult(1);
+
+        verify(onUnsubscribe).call();
+    }
+
+    @Test
+    public void v1ObservableIsUnsubscribedOnError2() {
+        Action0 onUnsubscribe = mock(Action0.class);
+
+        RxJavaInterop.toV2Flowable(rx.Observable.error(new IOException())
+        .doOnUnsubscribe(onUnsubscribe))
+        .test()
+        .assertFailure(IOException.class);
+
+        verify(onUnsubscribe).call();
+    }
+
+    @Test
+    public void v1ObservableIsUnsubscribedOnCompletion2() {
+        Action0 onUnsubscribe = mock(Action0.class);
+
+        RxJavaInterop.toV2Flowable(rx.Observable.just(1)
+        .doOnUnsubscribe(onUnsubscribe))
+        .test()
+        .assertResult(1);
+
+        verify(onUnsubscribe).call();
+    }
 }