UndeliverableException in ScanOperationApi21

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/operations/LegacyScanOperation.java

@@ -17,7 +17,7 @@
 import java.util.Set;
 import java.util.UUID;
 
-import io.reactivex.Emitter;
+import io.reactivex.ObservableEmitter;
 
 public class LegacyScanOperation extends ScanOperation<RxBleInternalScanResultLegacy, BluetoothAdapter.LeScanCallback> {
 
@@ -38,7 +38,7 @@ public LegacyScanOperation(UUID[] filterServiceUUIDs, RxBleAdapterWrapper rxBleA
     }
 
     @Override
-    BluetoothAdapter.LeScanCallback createScanCallback(final Emitter<RxBleInternalScanResultLegacy> emitter) {
+    BluetoothAdapter.LeScanCallback createScanCallback(final ObservableEmitter<RxBleInternalScanResultLegacy> emitter) {
         return new BluetoothAdapter.LeScanCallback() {
             @Override
             public void onLeScan(BluetoothDevice device, int rssi, byte[] scanRecord) {

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/operations/ScanOperation.java

@@ -9,7 +9,6 @@
 import com.polidea.rxandroidble2.internal.serialization.QueueReleaseInterface;
 import com.polidea.rxandroidble2.internal.util.RxBleAdapterWrapper;
 
-import io.reactivex.Emitter;
 import io.reactivex.ObservableEmitter;
 import io.reactivex.functions.Cancellable;
 
@@ -67,21 +66,21 @@ protected BleException provideException(DeadObjectException deadObjectException)
      * @return the scan callback type to use with {@link #startScan(RxBleAdapterWrapper, Object)}
      * and {@link #stopScan(RxBleAdapterWrapper, Object)}
      */
-    abstract SCAN_CALLBACK_TYPE createScanCallback(Emitter<SCAN_RESULT_TYPE> emitter);
+    abstract SCAN_CALLBACK_TYPE createScanCallback(ObservableEmitter<SCAN_RESULT_TYPE> emitter);
 
     /**
      * Function that should start the scan using passed {@link RxBleAdapterWrapper} and {@link SCAN_CALLBACK_TYPE} callback
      * @param rxBleAdapterWrapper the {@link RxBleAdapterWrapper} to use
-     * @param scanCallback the {@link SCAN_CALLBACK_TYPE} returned by {@link #createScanCallback(Emitter)} to start
+     * @param scanCallback the {@link SCAN_CALLBACK_TYPE} returned by {@link #createScanCallback(ObservableEmitter)} to start
      * @return true if successful
      */
     abstract boolean startScan(RxBleAdapterWrapper rxBleAdapterWrapper, SCAN_CALLBACK_TYPE scanCallback);
 
     /**
      * Method that should stop the scan for a given {@link SCAN_CALLBACK_TYPE} that was previously returned
-     * by {@link #createScanCallback(Emitter)}
+     * by {@link #createScanCallback(ObservableEmitter)}
      * @param rxBleAdapterWrapper the {@link RxBleAdapterWrapper} to use
-     * @param scanCallback the {@link SCAN_CALLBACK_TYPE} returned by {@link #createScanCallback(Emitter)} to stop
+     * @param scanCallback the {@link SCAN_CALLBACK_TYPE} returned by {@link #createScanCallback(ObservableEmitter)} to stop
      */
     abstract void stopScan(RxBleAdapterWrapper rxBleAdapterWrapper, SCAN_CALLBACK_TYPE scanCallback);
 }

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/operations/ScanOperationApi18.java

@@ -13,7 +13,7 @@
 import com.polidea.rxandroidble2.internal.scan.RxBleInternalScanResult;
 import com.polidea.rxandroidble2.internal.util.RxBleAdapterWrapper;
 
-import io.reactivex.Emitter;
+import io.reactivex.ObservableEmitter;
 
 public class ScanOperationApi18 extends ScanOperation<RxBleInternalScanResult, BluetoothAdapter.LeScanCallback> {
 
@@ -34,7 +34,7 @@ public ScanOperationApi18(
     }
 
     @Override
-    BluetoothAdapter.LeScanCallback createScanCallback(final Emitter<RxBleInternalScanResult> emitter) {
+    BluetoothAdapter.LeScanCallback createScanCallback(final ObservableEmitter<RxBleInternalScanResult> emitter) {
         return new BluetoothAdapter.LeScanCallback() {
             @Override
             public void onLeScan(BluetoothDevice device, int rssi, byte[] scanRecord) {

rxandroidble/src/main/java/com/polidea/rxandroidble2/internal/operations/ScanOperationApi21.java

@@ -23,7 +23,7 @@
 import java.util.Arrays;
 import java.util.List;
 
-import io.reactivex.Emitter;
+import io.reactivex.ObservableEmitter;
 
 @RequiresApi(21 /* Build.VERSION_CODES.LOLLIPOP */)
 public class ScanOperationApi21 extends ScanOperation<RxBleInternalScanResult, ScanCallback> {
@@ -57,7 +57,7 @@ public ScanOperationApi21(
     }
 
     @Override
-    ScanCallback createScanCallback(final Emitter<RxBleInternalScanResult> emitter) {
+    ScanCallback createScanCallback(final ObservableEmitter<RxBleInternalScanResult> emitter) {
         return new ScanCallback() {
             @Override
             public void onScanResult(int callbackType, ScanResult result) {
@@ -90,7 +90,7 @@ public void onBatchScanResults(List<ScanResult> results) {
 
             @Override
             public void onScanFailed(int errorCode) {
-                emitter.onError(new BleScanException(errorCodeToBleErrorCode(errorCode)));
+                emitter.tryOnError(new BleScanException(errorCodeToBleErrorCode(errorCode)));
             }
         };
     }

rxandroidble/src/test/groovy/com/polidea/rxandroidble2/internal/operations/OperationScanApi21Test.groovy

@@ -14,6 +14,7 @@ import com.polidea.rxandroidble2.internal.serialization.QueueReleaseInterface
 import com.polidea.rxandroidble2.internal.util.RxBleAdapterWrapper
 import com.polidea.rxandroidble2.scan.ScanFilter
 import com.polidea.rxandroidble2.scan.ScanSettings
+import io.reactivex.plugins.RxJavaPlugins
 import spock.lang.Specification
 import spock.lang.Unroll
 
@@ -184,6 +185,26 @@ public class OperationScanApi21Test extends Specification {
         [true, true]         | 2
     }
 
+    def "onScanFailed() should not result in exception being passed to RxJavaPlugins.onErrorHandler() if Observable was disposed."() {
+
+        given:
+        def capturedLeScanCallbackRef = captureScanCallback()
+        def rxUnhandledExceptionRef = captureRxUnhandledExceptions()
+        prepareObjectUnderTest(Mock(ScanSettings), null, null)
+        def testSubscriber = objectUnderTest.run(mockQueueReleaseInterface).test()
+        testSubscriber.dispose()
+
+        when:
+        capturedLeScanCallbackRef.get().onScanFailed(ScanCallback.SCAN_FAILED_APPLICATION_REGISTRATION_FAILED)
+
+        then:
+        rxUnhandledExceptionRef.get() == null
+
+        cleanup:
+        RxJavaPlugins.setErrorHandler(null)
+
+    }
+
     private AtomicReference<ScanCallback> captureScanCallback() {
         AtomicReference<ScanCallback> scanCallbackAtomicReference = new AtomicReference<>()
         mockAdapterWrapper.startLeScan(_, _, _) >> { List<ScanFilter> _, ScanSettings _1, ScanCallback scanCallback ->
@@ -192,6 +213,14 @@ public class OperationScanApi21Test extends Specification {
         return scanCallbackAtomicReference
     }
 
+    private static AtomicReference<Throwable> captureRxUnhandledExceptions() {
+        AtomicReference<Throwable> unhandledExceptionAtomicReference = new AtomicReference<>()
+        RxJavaPlugins.setErrorHandler({ throwable ->
+            unhandledExceptionAtomicReference.set(throwable)
+        })
+        return unhandledExceptionAtomicReference
+    }
+
     private static class MockBleAdapterWrapper extends RxBleAdapterWrapper {
 
         Semaphore acquireBeforeReturnStartScan = null