RxAndroidBle stopped when discovering another BLE device

Question

I'm using RxAndroidBle to discovery surrounded devices' service and some characteristics. I met a problem that establishConnection with another device dosn't work after the connection with first device is unsubsribed. When it comes to another device, the state is always "Connection State: RxBleConnectionState{DISCONNECTED}".

Here is the codes, will anyone can help me to have a check!

private Subscription createDeviceConnectionSubscription(final RxBleDevice 
rxBleDevice){
    bInternalStatus = DISCOVERRING;
    Log.d(LOG_TAG, "start to discover the services on: " + rxBleDevice.getMacAddress().toString());

    Subscription subscription = rxBleDevice.establishConnection(false)
            .timeout(10, TimeUnit.SECONDS)
            .subscribeOn(AndroidSchedulers.mainThread())
            .observeOn(AndroidSchedulers.from(mWorker.getLooper()))
            .flatMap(new Func1<RxBleConnection, Observable<gattConnectionResults>>() {
                @Override
                public Observable<gattConnectionResults> call(RxBleConnection rxBleConnection) {
                    return Observable.zip(
                            rxBleConnection.discoverServices(3, TimeUnit.SECONDS),               
               rxBleConnection.readCharacteristic(UUID.fromString("UUID1")),                                
               rxBleConnection.readCharacteristic(UUID.fromString("UUID2")),
                            new Func3<RxBleDeviceServices, byte[], byte[], gattConnectionResults>() {
                                @Override
                                public gattConnectionResults call(RxBleDeviceServices rxBleDeviceServices, byte[] bytes, byte[] bytes2) {
                                    //......
                                    return results;
                                }
                            }
                    );
                }
            })
            .subscribe(subscriber);

    rxBleDevice.observeConnectionStateChanges()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<RxBleConnection.RxBleConnectionState>() {
                @Override
                public void call(RxBleConnection.RxBleConnectionState rxBleConnectionState) {
                    Log.d("Gatt Connection State", ""+rxBleConnectionState);
                }
            });
 return subscription;
}

Subscriber<gattConnectionResults> subscriber = new 
Subscriber<gattConnectionResults>() {
    @Override
    public void onCompleted() {
        Log.d(LOG_TAG, "RXXX on completed called!!!");
        if(mDeviceConnectionSubscription!=null){
            handleConnectionOver();
        }
    }

    @Override
    public void onError(Throwable e) {
        Log.d(LOG_TAG, "RXXX Connection error!!! " + e);
        if(mDeviceConnectionSubscription!=null){
            handleConnectionOver();
        }
    }

    @Override
    public void onNext(gattConnectionResults results) {

        }
    }
};

private void handleConnectionOver(){
    bInternalStatus = STOPPED;
    if(mDeviceConnectionSubscription!=null && !mDeviceConnectionSubscription.isUnsubscribed()){
        mDeviceConnectionSubscription.unsubscribe();
        mDeviceConnectionSubscription = null;
    }
}

Thanks for @s-noopy, after move unsubcribe to the end of onNext(), the logs by RxBle are:

05-15 D/RxBle#Radio:   QUEUED RxBleRadioOperationConnect(192750313)
05-15 D/RxBle#Radio:  STARTED RxBleRadioOperationConnect(192750313)
05-15 V/RxBle#BleConnectionCompat: Connecting without reflection
05-15 D/RxBle#BluetoothGatt: onConnectionStateChange newState=2 status=0
05-15 D/RxBle#Radio: FINISHED RxBleRadioOperationConnect(192750313)
05-15 D/RxBle#Radio:   QUEUED RxBleRadioOperationServicesDiscover(178598822)
05-15 D/RxBle#Radio:  STARTED RxBleRadioOperationServicesDiscover(178598822)
05-15 D/RxBle#BluetoothGatt: onServicesDiscovered status=0
05-15 D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicRead(241700341)
05-15 D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicRead(24572440)
05-15 D/RxBle#Radio: FINISHED RxBleRadioOperationServicesDiscover(178598822)
05-15 D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicRead(241700341)
05-15 D/RxBle#BluetoothGatt: onCharacteristicRead characteristic=00002a00-0000-1000-8000-00805f9b34fb status=0
05-15 D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicRead(241700341)
05-15 D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicRead(24572440)
05-15 D/RxBle#BluetoothGatt: onCharacteristicRead characteristic=00002a29-0000-1000-8000-00805f9b34fb status=0
05-15 D/RxBle#Radio:   QUEUED RxBleRadioOperationDisconnect(204169407)
05-15 D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicRead(24572440)
05-15 D/RxBle#Radio:  STARTED RxBleRadioOperationDisconnect(204169407)
05-15 D/RxBle#BluetoothGatt: onConnectionStateChange newState=0 status=0
05-15 D/RxBle#Radio: FINISHED RxBleRadioOperationDisconnect(204169407)

Answer 1

You should create a new Subscriber for each Subscription. Subscriber is stateful and once it will get onCompleted or onError it will not subscribe again to any other Observable.

Alternatively you should use Observer instead of Subscriber. Observer is stateless.