0

How can I perform blocking write operation in Android with RxAndroidBle. Only if the write operation is successful the next command should be performed.

protected void doWriteBytes(UUID characteristic, byte[] bytes) {
    final Disposable disposable = connectionObservable
            .flatMapSingle(rxBleConnection -> rxBleConnection.writeCharacteristic(characteristic, bytes))
            .observeOn(AndroidSchedulers.mainThread())
            .retry(BT_RETRY_TIMES_ON_ERROR)
            .subscribe(
                    value -> {
                        Timber.d("Write characteristic %s: %s",
                                BluetoothGattUuid.prettyPrint(characteristic),
                                byteInHex(value));
                    },
                    throwable -> onError(throwable)
            );

    compositeDisposable.add(disposable);
}


protected void test() {
  // blocking write bytes
  doWriteBytes(UUID.fromString("f433bd80-75b8-11e2-97d9-0002a5d5c51b"), new byte[] {0x35, 0x12});

  // following command should be only performed if doWriteBytes is successful executed
  foo();

  // blocking write bytes
  doWriteBytes(UUID.fromString("f433bd80-75b8-11e2-97d9-0002a5d5c51b"), new byte[] {0x5, 0x6, 0x1});

  bar();
}

I know subscribe and onComplete but it is also possible to do without these methods?

The background is I want to override the test method in several different subclasses, so I can perform various doWriteBytes commands (for e.g. ACK commands) to send some bytes to a Bluetooth device but I need to be sure that the next command is only performed if the ACK command is successful send.

Maybe it is more a RxJava2 problem but I am not quite familiar with it.

Edit:

Thanks for your answer @Dariusz Seweryn. Sorry my question was probably not really clear. I will try to concretize it.

I want to write the source code like a normal function in test() to abstract the RxJava2 implementations. The only different is that the doWriteBytes and other Bluetooth operations (notification, read) should be done via RxAndroidBle. What I have to write to a Bluetooth device depends on the notifications bytes or some other algorithm in the test() method. Additionally, I want to overwrite the test() method to implement a different Bluetooth communication flow for completely different Bluetooth device. It is always important that the Bluetooth operations are sequentially processed.

Now I have three ideas in mind:

1) My first idea is to implement all RxAndroidBle operations blocking, so I can use simple for e.g. loops.

2) My second idea is to dynamically add (concat?) at runtime observations to another in the test() method which is sequentially processes but I need always the return values from the previous observations?

3) My third idea is to combine write/notify/write operation as a method, which I can call in the test() method. The operation should be write bytes to a characterstic A then wait for the notification on characteristic B, do some processing with the received bytes and writes again to characteristic C. But what is written or how the notification process should be dynamically during runtime in the test() method added.

Maybe there is a elegant solution for my problem in RxJava2 or it is not possible at all?

Edit2:

I tried to implement all three ideas but unfortunately I didn't succed.

1)

connectionObservable
                .flatMapSingle(rxBleConnection -> rxBleConnection.writeCharacteristic(characteristic, bytes))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .retry(BT_RETRY_TIMES_ON_ERROR)
                .blockingSubscribe(
                        value -> {
                            Timber.d("Write characteristic %s: %s",
                                    BluetoothGattUuid.prettyPrint(characteristic),
                                    byteInHex(value));
                            processBtQueue();
                        },
                        throwable -> onError(throwable)
                );

It is always blocking even on success? Do I have to release it somewhere? Additionally the method return void and not a disposable anymore but then I can't dispose it.

2) I am struggling at this idea. On which observable should I concat to if I don't know the starting observable? The connectionObserable doesn't work because it holds the RxBleConnection. The second problem is that the values after a Bluetooth operation are then Java Object classes!? Do I have to cast it every time? Do you have an example how I can concat for example a Bluetooth write operation to a notification Bluetooth result?

3) The problem with this idea is that I don't know how to dynamical add at runtime the processing part to the notification outside of the RxJava subscribe part?

I have a working solution for idea nr 3

protected Observable<byte[]> doWriteNotify(UUID characteristic, byte[] bytes, Observable<byte[]> notificationObservable) {
    Observable observable = connectionObservable
            .flatMapSingle(rxBleConnection -> rxBleConnection.writeCharacteristic(characteristic, bytes))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .retry(BT_RETRY_TIMES_ON_ERROR)
        .flatMap( writeBytes -> notificationObservable)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .retry(BT_RETRY_TIMES_ON_ERROR);

    compositeDisposable.add(observable.subscribe());

    return observable;
}

Btw. should I create separate threads on stackoverflow with these questions?

If it's help, you can find my experimental source code here.

2
  • Keep in mind that editing your post does not notify people who have answered already Feb 20, 2019 at 15:23
  • Thanks for your hint. I edited my question again.
    – olie.xdev
    Feb 20, 2019 at 16:14

1 Answer 1

0

I know subscribe and onComplete but it is also possible to do without these methods?

Given:

Completable foo() { ... }

Completable bar() { ... )

One could do:

Disposable testDisposable = connectionObservable
                                .flatMapCompletable(connection ->
                                    connection.writeCharacteristic(UUID.fromString("f433bd80-75b8-11e2-97d9-0002a5d5c51b"), new byte[] {0x35, 0x12}).ignoreElement()
                                        .andThen(foo())
                                        .andThen(connection.writeCharacteristic(UUID.fromString("f433bd80-75b8-11e2-97d9-0002a5d5c51b"), new byte[] {0x5, 0x6, 0x1}).ignoreElement())
                                        .andThen(bar())
                                )
                                .subscribe(
                                     () -> { /* completed */ },
                                     throwable -> { /* error happened */ }
                                )

Having the above closed with one .subscribe() one can have control over how many connections the flow will finish. In the above example if a connection would end prematurely during the first write — all next operations (foo(), write, bar()) will not happen at all.

Edit:

All your ideas could potentially work — you can try them out.

Maybe there is an elegant solution for my problem in RxJava2 or it is not possible at all?

There are .blocking*() functions for Observable/Single/Completable classes if you really need them. Be careful though — you may start experiencing some hard to debug issues depending on your implementation due to introducing more state to your application.

3
  • Thanks for your help. I resolved my problem with the third idea and adding subscribeOn(Schedulers.io()) but I would still be interested how I resolve the blockingSubscribe()?
    – olie.xdev
    Feb 20, 2019 at 20:33
  • Not .blockingSubscribe() but .blockingFirst() for instance. It would return the first emission in a blocking manner. I am not sure though how each of these functions handle disposing. Feb 21, 2019 at 9:51
  • hmmm.... won't really work as you expected. I having trouble with dead locks with blockingFirst. Anyway thanks for your great open-source RxAndroidBle library.
    – olie.xdev
    Feb 21, 2019 at 17:44

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.