Thread calling the Native API changed from main to background. (#255 #41

)

Initially when working with the Android Bluetooth API (18 / Android 4.3) it was found that specific implementations (Samsung’s) cannot work correctly if called directly in callbacks from other calls. To mitigate the problem the next interaction was routed to Android main thread. Now when investigating the issue it turned out that it may be any other thread as long as it is not the callback thread.

rxandroidble/build.gradle

@@ -46,7 +46,6 @@ android {
 
 dependencies {
     compile rootProject.ext.libs.rxjava
-    compile rootProject.ext.libs.rxandroid
     compile rootProject.ext.libs.rxrelay
     compile rootProject.ext.libs.support_annotations
     compile rootProject.ext.libs.dagger

rxandroidble/src/main/java/com/polidea/rxandroidble/ClientComponent.java

@@ -9,6 +9,7 @@
 import android.location.LocationManager;
 import android.os.Build;
 import android.support.annotation.Nullable;
+import android.support.annotation.RestrictTo;
 import com.polidea.rxandroidble.helpers.LocationServicesOkObservable;
 import com.polidea.rxandroidble.internal.DeviceComponent;
 import com.polidea.rxandroidble.internal.scan.InternalToExternalScanResultConverter;
@@ -33,7 +34,6 @@
 import javax.inject.Provider;
 import rx.Observable;
 import rx.Scheduler;
-import rx.android.schedulers.AndroidSchedulers;
 import rx.functions.Func1;
 import rx.schedulers.Schedulers;
 
@@ -43,11 +43,10 @@ public interface ClientComponent {
 
     class NamedSchedulers {
 
-        public static final String MAIN_THREAD = "main-thread";
         public static final String COMPUTATION = "computation";
-        public static final String RADIO_OPERATIONS = "callback-emitter";
         public static final String TIMEOUT = "timeout";
-        public static final String GATT_CALLBACK = "callback";
+        public static final String BLUETOOTH_INTERACTION = "bluetooth_interaction";
+        public static final String BLUETOOTH_CALLBACKS = "bluetooth_callbacks";
         private NamedSchedulers() {
 
         }
@@ -119,28 +118,50 @@ static int provideDeviceSdk() {
         }
 
         @Provides
-        @Named(NamedSchedulers.GATT_CALLBACK)
+        @Named(NamedSchedulers.BLUETOOTH_INTERACTION)
         @ClientScope
-        static ExecutorService provideGattCallbackExecutorService() {
+        static ExecutorService provideBluetoothInteractionExecutorService() {
             return Executors.newSingleThreadExecutor();
         }
 
         @Provides
-        @Named(NamedSchedulers.GATT_CALLBACK)
+        @Named(NamedSchedulers.BLUETOOTH_CALLBACKS)
         @ClientScope
-        static Scheduler provideGattCallbackScheduler(@Named(NamedSchedulers.GATT_CALLBACK) ExecutorService executorService) {
-            return Schedulers.from(executorService);
+        static ExecutorService provideBluetoothCallbacksExecutorService() {
+            return Executors.newSingleThreadExecutor();
         }
 
         @Provides
-        LocationManager provideLocationManager() {
-            return (LocationManager) context.getSystemService(Context.LOCATION_SERVICE);
+        @Named(NamedSchedulers.BLUETOOTH_INTERACTION)
+        @ClientScope
+        static Scheduler provideBluetoothInteractionScheduler(@Named(NamedSchedulers.BLUETOOTH_INTERACTION) ExecutorService service) {
+            return Schedulers.from(service);
+        }
+
+        @Provides
+        @Named(NamedSchedulers.BLUETOOTH_CALLBACKS)
+        @ClientScope
+        static Scheduler provideBluetoothCallbacksScheduler(@Named(NamedSchedulers.BLUETOOTH_CALLBACKS) ExecutorService service) {
+            return Schedulers.from(service);
         }
 
         @Provides
-        @Named(NamedSchedulers.MAIN_THREAD)
-        static Scheduler provideMainThreadScheduler() {
-            return AndroidSchedulers.mainThread();
+        static ClientComponentFinalizer provideFinalizationCloseable(
+                @Named(NamedSchedulers.BLUETOOTH_INTERACTION) final ExecutorService interactionExecutorService,
+                @Named(NamedSchedulers.BLUETOOTH_CALLBACKS) final ExecutorService callbacksExecutorService
+        ) {
+            return new ClientComponentFinalizer() {
+                @Override
+                public void onFinalize() {
+                    interactionExecutorService.shutdown();
+                    callbacksExecutorService.shutdown();
+                }
+            };
+        }
+
+        @Provides
+        LocationManager provideLocationManager() {
+            return (LocationManager) context.getSystemService(Context.LOCATION_SERVICE);
         }
 
         @Provides
@@ -227,10 +248,6 @@ abstract class ClientModuleBinder {
         @ClientScope
         abstract RxBleRadio bindRxBleRadio(RxBleRadioImpl rxBleRadio);
 
-        @Binds
-        @Named(NamedSchedulers.RADIO_OPERATIONS)
-        abstract Scheduler bindCallbackScheduler(@Named(NamedSchedulers.MAIN_THREAD) Scheduler mainThreadScheduler);
-
         @Binds
         @Named(NamedSchedulers.TIMEOUT)
         abstract Scheduler bindTimeoutScheduler(@Named(NamedSchedulers.COMPUTATION) Scheduler computationScheduler);
@@ -242,4 +259,10 @@ abstract class ClientModuleBinder {
     LocationServicesOkObservable locationServicesOkObservable();
 
     RxBleClient rxBleClient();
+
+    @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
+    interface ClientComponentFinalizer {
+
+        void onFinalize();
+    }
 }

rxandroidble/src/main/java/com/polidea/rxandroidble/RxBleClientImpl.java

@@ -31,7 +31,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.ExecutorService;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -50,8 +49,8 @@ class RxBleClientImpl extends RxBleClient {
     private final ScanSetupBuilder scanSetupBuilder;
     private final ScanPreconditionsVerifier scanPreconditionVerifier;
     private final Func1<RxBleInternalScanResult, ScanResult> internalToExternalScanResultMapFunction;
-    private final ExecutorService executorService;
-    private final Scheduler mainThreadScheduler;
+    private final ClientComponent.ClientComponentFinalizer clientComponentFinalizer;
+    private final Scheduler bluetoothInteractionScheduler;
     private final Map<Set<UUID>, Observable<RxBleScanResult>> queuedScanOperations = new HashMap<>();
     private final RxBleAdapterWrapper rxBleAdapterWrapper;
     private final Observable<BleAdapterState> rxBleAdapterStateObservable;
@@ -69,8 +68,8 @@ class RxBleClientImpl extends RxBleClient {
                     ScanSetupBuilder scanSetupBuilder,
                     ScanPreconditionsVerifier scanPreconditionVerifier,
                     Func1<RxBleInternalScanResult, ScanResult> internalToExternalScanResultMapFunction,
-                    @Named(ClientComponent.NamedSchedulers.GATT_CALLBACK) ExecutorService executorService,
-                    @Named(ClientComponent.NamedSchedulers.MAIN_THREAD) Scheduler mainThreadScheduler) {
+                    @Named(ClientComponent.NamedSchedulers.BLUETOOTH_INTERACTION) Scheduler bluetoothInteractionScheduler,
+                    ClientComponent.ClientComponentFinalizer clientComponentFinalizer) {
         this.uuidUtil = uuidUtil;
         this.rxBleRadio = rxBleRadio;
         this.rxBleAdapterWrapper = rxBleAdapterWrapper;
@@ -81,14 +80,14 @@ class RxBleClientImpl extends RxBleClient {
         this.scanSetupBuilder = scanSetupBuilder;
         this.scanPreconditionVerifier = scanPreconditionVerifier;
         this.internalToExternalScanResultMapFunction = internalToExternalScanResultMapFunction;
-        this.executorService = executorService;
-        this.mainThreadScheduler = mainThreadScheduler;
+        this.bluetoothInteractionScheduler = bluetoothInteractionScheduler;
+        this.clientComponentFinalizer = clientComponentFinalizer;
     }
 
     @Override
     protected void finalize() throws Throwable {
+        clientComponentFinalizer.onFinalize();
         super.finalize();
-        executorService.shutdown();
     }
 
     @Override
@@ -118,7 +117,7 @@ public Observable<ScanResult> call() {
                 final ScanSetup scanSetup = scanSetupBuilder.build(scanSettings, scanFilters);
                 final Operation<RxBleInternalScanResult> scanOperation = scanSetup.scanOperation;
                 return rxBleRadio.queue(scanOperation)
-                        .unsubscribeOn(mainThreadScheduler)
+                        .unsubscribeOn(bluetoothInteractionScheduler)
                         .compose(scanSetup.scanOperationBehaviourEmulatorTransformer)
                         .map(internalToExternalScanResultMapFunction)
                         .mergeWith(RxBleClientImpl.this.<ScanResult>bluetoothAdapterOffExceptionObservable());

rxandroidble/src/main/java/com/polidea/rxandroidble/RxBleRadioOperationCustom.java

@@ -9,7 +9,6 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Scheduler;
-import rx.android.schedulers.AndroidSchedulers;
 
 /**
  * Represents a custom operation that will be enqueued for future execution within the client instance.
@@ -41,7 +40,6 @@ public interface RxBleRadioOperationCustom<T> {
      * @param bluetoothGatt     The Android API GATT instance
      * @param rxBleGattCallback The internal Rx ready bluetooth gatt callback to be notified of GATT operations
      * @param scheduler         The RxBleRadio scheduler used to asObservable operation
-     *                          (currently {@link AndroidSchedulers#mainThread()}
      * @throws Throwable Any exception that your custom operation might throw
      */
     @NonNull

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/connection/RxBleConnectionImpl.java

@@ -70,8 +70,9 @@ public RxBleConnectionImpl(
             DescriptorWriter descriptorWriter,
             OperationsProvider operationProvider,
             Provider<LongWriteOperationBuilder> longWriteOperationBuilderProvider,
-            @Named(ClientComponent.NamedSchedulers.RADIO_OPERATIONS) Scheduler callbackScheduler,
-            IllegalOperationChecker illegalOperationChecker) {
+            @Named(ClientComponent.NamedSchedulers.BLUETOOTH_INTERACTION) Scheduler callbackScheduler,
+            IllegalOperationChecker illegalOperationChecker
+    ) {
         this.rxBleRadio = rxBleRadio;
         this.gattCallback = gattCallback;
         this.bluetoothGatt = bluetoothGatt;

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/connection/RxBleGattCallback.java

@@ -69,7 +69,7 @@ public Object call(Pair<BluetoothGatt, RxBleConnectionState> bluetoothGattRxBleC
             .autoConnect(0);
 
     @Inject
-    public RxBleGattCallback(@Named(ClientComponent.NamedSchedulers.GATT_CALLBACK) Scheduler callbackScheduler,
+    public RxBleGattCallback(@Named(ClientComponent.NamedSchedulers.BLUETOOTH_CALLBACKS) Scheduler callbackScheduler,
                              BluetoothGattProvider bluetoothGattProvider,
                              NativeCallbackDispatcher nativeCallbackDispatcher) {
         this.callbackScheduler = callbackScheduler;

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/operations/OperationsProviderImpl.java

@@ -25,7 +25,7 @@ public class OperationsProviderImpl implements OperationsProvider {
     private final RxBleGattCallback rxBleGattCallback;
     private final BluetoothGatt bluetoothGatt;
     private final TimeoutConfiguration timeoutConfiguration;
-    private final Scheduler mainThreadScheduler;
+    private final Scheduler bluetoothInteractionScheduler;
     private final Scheduler timeoutScheduler;
     private final Provider<RxBleRadioOperationReadRssi> rssiReadOperationProvider;
 
@@ -34,13 +34,13 @@ public class OperationsProviderImpl implements OperationsProvider {
             RxBleGattCallback rxBleGattCallback,
             BluetoothGatt bluetoothGatt,
             @Named(DeviceModule.OPERATION_TIMEOUT) TimeoutConfiguration timeoutConfiguration,
-            @Named(ClientComponent.NamedSchedulers.MAIN_THREAD) Scheduler mainThreadScheduler,
+            @Named(ClientComponent.NamedSchedulers.BLUETOOTH_INTERACTION) Scheduler bluetoothInteractionScheduler,
             @Named(ClientComponent.NamedSchedulers.TIMEOUT) Scheduler timeoutScheduler,
             Provider<RxBleRadioOperationReadRssi> rssiReadOperationProvider) {
         this.rxBleGattCallback = rxBleGattCallback;
         this.bluetoothGatt = bluetoothGatt;
         this.timeoutConfiguration = timeoutConfiguration;
-        this.mainThreadScheduler = mainThreadScheduler;
+        this.bluetoothInteractionScheduler = bluetoothInteractionScheduler;
         this.timeoutScheduler = timeoutScheduler;
         this.rssiReadOperationProvider = rssiReadOperationProvider;
     }
@@ -54,7 +54,7 @@ public RxBleRadioOperationCharacteristicLongWrite provideLongWriteOperation(
 
         return new RxBleRadioOperationCharacteristicLongWrite(bluetoothGatt,
                 rxBleGattCallback,
-                mainThreadScheduler,
+                bluetoothInteractionScheduler,
                 timeoutConfiguration,
                 bluetoothGattCharacteristic,
                 maxBatchSizeProvider,

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/operations/RxBleRadioOperationCharacteristicLongWrite.java

@@ -38,7 +38,7 @@ public class RxBleRadioOperationCharacteristicLongWrite extends RxBleRadioOperat
 
     private final BluetoothGatt bluetoothGatt;
     private final RxBleGattCallback rxBleGattCallback;
-    private final Scheduler mainThreadScheduler;
+    private final Scheduler bluetoothInteractionScheduler;
     private final TimeoutConfiguration timeoutConfiguration;
     private final BluetoothGattCharacteristic bluetoothGattCharacteristic;
     private final PayloadSizeLimitProvider batchSizeProvider;
@@ -49,15 +49,15 @@ public class RxBleRadioOperationCharacteristicLongWrite extends RxBleRadioOperat
     RxBleRadioOperationCharacteristicLongWrite(
             BluetoothGatt bluetoothGatt,
             RxBleGattCallback rxBleGattCallback,
-            @Named(ClientComponent.NamedSchedulers.MAIN_THREAD) Scheduler mainThreadScheduler,
+            @Named(ClientComponent.NamedSchedulers.BLUETOOTH_INTERACTION) Scheduler bluetoothInteractionScheduler,
             @Named(DeviceModule.OPERATION_TIMEOUT) TimeoutConfiguration timeoutConfiguration,
             BluetoothGattCharacteristic bluetoothGattCharacteristic,
             PayloadSizeLimitProvider batchSizeProvider,
             WriteOperationAckStrategy writeOperationAckStrategy,
             byte[] bytesToWrite) {
         this.bluetoothGatt = bluetoothGatt;
         this.rxBleGattCallback = rxBleGattCallback;
-        this.mainThreadScheduler = mainThreadScheduler;
+        this.bluetoothInteractionScheduler = bluetoothInteractionScheduler;
         this.timeoutConfiguration = timeoutConfiguration;
         this.bluetoothGattCharacteristic = bluetoothGattCharacteristic;
         this.batchSizeProvider = batchSizeProvider;
@@ -79,7 +79,7 @@ protected void protectedRun(final Emitter<byte[]> emitter, final RadioReleaseInt
 
         final RadioReleasingEmitterWrapper<byte[]> emitterWrapper = new RadioReleasingEmitterWrapper<>(emitter, radioReleaseInterface);
         writeBatchAndObserve(batchSize, byteBuffer)
-                .subscribeOn(mainThreadScheduler)
+                .subscribeOn(bluetoothInteractionScheduler)
                 .takeFirst(writeResponseForMatchingCharacteristic(bluetoothGattCharacteristic))
                 .timeout(
                         timeoutConfiguration.timeout,