RxAndroidBleでConnectを行いたい

Question

RxAndroidBleでスキャンする
の記事を参考にスキャンを行い身の回りにあるデバイスの名前を出力することができました。

次にconnectを行いたいと思ったのですがこれがどう実装すれば良いのかわかりません。
READMEには以下の様にconnectすると記述されていました。

String macAddress = "AA:BB:CC:DD:EE:FF";
RxBleDevice device = rxBleClient.getBleDevice(macAddress);

Disposable disposable = device.establishConnection(false) // <-- autoConnect flag
    .subscribe(
        rxBleConnection -> {
            // All GATT operations are done through the rxBleConnection.
        },
        throwable -> {
            // Handle an error here.
        }
    );

// When done... dispose and forget about connection teardown :)
disposable.dispose();

現在コードは以下の様になっています。
establishConnectionでConnectをするとREADMEにあったので以下のように試してみたのですがConnectできません。何か別途必要なものがあるのでしょうか?

val scanSubscription = rxBleClient.scanBleDevices(
            ScanSettings.Builder().build()
        ).filter {
            it.scanRecord.deviceName != null

        }.subscribe({scanResult ->
            Log.i("ScanDevice:",scanResult.scanRecord.deviceName)
            val cubeAdress = scanResult.bleDevice.macAddress
            //ここでconnectしたい
            scanResult.bleDevice.establishConnection(false)
            
        },{throwable ->
            Log.i("error","$throwable")
        })

Answer 1

READMEによれば、RxBleDevice#establishConnectionはcoldなObservableを返すメソッドのようですから、subscribeを呼んであげなければ目的の処理が実行されることはないはずです。

switchMap等でObservableをチェーンするようにしてみてください。

val disposable = rxBleClient
    .scanBleDevices(ScanSettings.Builder().build())
    .filter { it.scanRecord.deviceName != null }
    .switchMap { it.bleDevice.establishConnection(false) }
    .subscribe({ rxBleConnection ->
        // All GATT operations are done through the rxBleConnection.
    }, { throwable ->
        // Handle an error here.
    })