Fixed RxBleAdapterStateObservable leak.

Author: dariuszseweryn

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/connection/DisconnectAction.java

@@ -5,18 +5,23 @@
 
 import bleshadow.javax.inject.Inject;
 
-import io.reactivex.internal.functions.Functions;
+import io.reactivex.functions.Action;
 
 @ConnectionScope
 class DisconnectAction implements ConnectionSubscriptionWatcher {
 
     private final ClientOperationQueue clientOperationQueue;
     private final DisconnectOperation operationDisconnect;
+    private final DisconnectionRouterInput disconnectionRouterInput;
 
     @Inject
-    DisconnectAction(ClientOperationQueue clientOperationQueue, DisconnectOperation operationDisconnect) {
+    DisconnectAction(
+            ClientOperationQueue clientOperationQueue,
+            DisconnectOperation operationDisconnect,
+            DisconnectionRouterInput disconnectionRouterInput) {
         this.clientOperationQueue = clientOperationQueue;
         this.operationDisconnect = operationDisconnect;
+        this.disconnectionRouterInput = disconnectionRouterInput;
     }
 
     @Override
@@ -28,9 +33,13 @@ public void onConnectionSubscribed() {
     public void onConnectionUnsubscribed() {
         clientOperationQueue
                 .queue(operationDisconnect)
-                .subscribe(
-                        Functions.emptyConsumer(),
-                        Functions.emptyConsumer()
-                );
+                .ignoreElements()
+                .onErrorComplete()
+                .subscribe(new Action() {
+                    @Override
+                    public void run() {
+                        disconnectionRouterInput.close();
+                    }
+                });
     }
 }

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/connection/DisconnectionRouter.java

@@ -1,7 +1,6 @@
 package com.polidea.rxandroidble2.internal.connection;
 
 
-import com.jakewharton.rxrelay2.BehaviorRelay;
 import com.polidea.rxandroidble2.RxBleAdapterStateObservable;
 import com.polidea.rxandroidble2.exceptions.BleDisconnectedException;
 import com.polidea.rxandroidble2.exceptions.BleException;
@@ -19,14 +18,15 @@
 import io.reactivex.functions.Consumer;
 import io.reactivex.functions.Function;
 import io.reactivex.functions.Predicate;
+import io.reactivex.subjects.BehaviorSubject;
 
 /**
  * A class that is responsible for routing all potential sources of disconnection to an Observable that emits only errors.
  */
 @ConnectionScope
 class DisconnectionRouter implements DisconnectionRouterInput, DisconnectionRouterOutput {
 
-    private final BehaviorRelay<BleException> bleExceptionBehaviorRelay = BehaviorRelay.create();
+    private final BehaviorSubject<BleException> bleExceptionBehaviorSubject = BehaviorSubject.create();
     private final Observable<BleException> firstDisconnectionValueObs;
     private final Observable<Object> firstDisconnectionExceptionObs;
 
@@ -59,14 +59,19 @@ public void accept(BleException exception) {
                         RxBleLog.v("An exception received, indicating that the adapter has became unusable.");
                     }
                 })
-                .subscribe(bleExceptionBehaviorRelay, new Consumer<Throwable>() {
+                .subscribe(new Consumer<BleException>() {
+                    @Override
+                    public void accept(BleException t) {
+                        bleExceptionBehaviorSubject.onNext(t);
+                    }
+                }, new Consumer<Throwable>() {
                     @Override
                     public void accept(Throwable throwable) {
                         RxBleLog.e(throwable, "Failed to monitor adapter state.");
                     }
                 });
 
-        firstDisconnectionValueObs = bleExceptionBehaviorRelay
+        firstDisconnectionValueObs = bleExceptionBehaviorSubject
                 .firstElement()
                 .toObservable()
                 .doOnTerminate(new Action() {
@@ -107,12 +112,17 @@ public boolean test(Boolean isAdapterUsable) {
 
     @Override
     public void onDisconnectedException(BleDisconnectedException disconnectedException) {
-        bleExceptionBehaviorRelay.accept(disconnectedException);
+        bleExceptionBehaviorSubject.onNext(disconnectedException);
     }
 
     @Override
     public void onGattConnectionStateException(BleGattException disconnectedGattException) {
-        bleExceptionBehaviorRelay.accept(disconnectedGattException);
+        bleExceptionBehaviorSubject.onNext(disconnectedGattException);
+    }
+
+    @Override
+    public void close() {
+        bleExceptionBehaviorSubject.onComplete();
     }
 
     @Override

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/connection/DisconnectionRouterInput.java

@@ -24,4 +24,9 @@ interface DisconnectionRouterInput {
      * @param disconnectedGattException the exception that happened
      */
     void onGattConnectionStateException(BleGattException disconnectedGattException);
+
+    /**
+     * Method to be called when no other calls are expected. Used for cleanup.
+     */
+    void close();
 }

rxandroidble/src/test/groovy/com/polidea/rxandroidble2/internal/connection/DisconnectionRouterTest.groovy

@@ -240,7 +240,7 @@ class DisconnectionRouterTest extends Specification {
     }
 
     @Unroll
-    def "should unsubscribe from adapterStateObservable if it emits STATE_OFF/STATE_TURNING_* or if .on*Exception() is called"() {
+    def "should unsubscribe from adapterStateObservable if it emits STATE_OFF/STATE_TURNING_* or .on*Exception()/.close() is called"() {
 
         given:
         createObjectUnderTest(true)
@@ -268,6 +268,9 @@ class DisconnectionRouterTest extends Specification {
                 { PublishSubject<RxBleAdapterStateObservable.BleAdapterState> mockAdapterStateSubject, DisconnectionRouter objectUnderTest ->
                     objectUnderTest.onDisconnectedException(new BleDisconnectedException("test"))
                 },
+                { PublishSubject<RxBleAdapterStateObservable.BleAdapterState> mockAdapterStateSubject, DisconnectionRouter objectUnderTest ->
+                    objectUnderTest.close()
+                },
         ]
     }
 }