一次发出一个项目，与它交互直到满足条件，然后继续下一个项目

Question

我有一个 BLE 设备列表，并且正在使用 RxJava 与它们进行交互。我需要从列表中发出一个项目，重复写入一个特征直到 X 发生，然后继续到列表中的下一个项目。

当前代码：

Observable.from(mDevices)
                .flatMap(new Func1<Device, Observable<?>>() {
                    @Override
                    public Observable<?> call(Device device) {
                        Log.d(TAG, "connecting for policing");
                        return device.connectForPolicing();
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {
                        Log.d(TAG, "subscribing... ");
                    }
                });

.connectForPolicing() 的样子：

public Observable<byte[]> connectForPolice() {

        ....

        return device.establishConnection(mContext, false)
                .flatMap(new Func1<RxBleConnection, Observable<byte[]>>() {
                    @Override
                    public Observable<byte[]> call(RxBleConnection rxBleConnection) {
                        byte[] value = new byte[1];
                        value[0] = (byte) (3 & 0xFF);
                        //Buzz the device
                        return rxBleConnection.writeCharacteristic(Constants.BUZZER_SELECT, value);
                    }
                })
                .repeat(3)//ignore
                .takeUntil(device.observeConnectionStateChanges().filter(new Func1<RxBleConnection.RxBleConnectionState, Boolean>() {
                    @Override
                    public Boolean call(RxBleConnection.RxBleConnectionState rxBleConnectionState) {

                        return rxBleConnectionState == RxBleConnection.RxBleConnectionState.DISCONNECTING;
                    }
                }));
    }

此代码似乎会立即发出列表中的所有项目，因此会同时连接和蜂鸣所有项目。如何一次发出一个项目以便与它们交互？

伪代码类似于：

for(Device device : devices) {
    device.connect();
    while(device.isConnected()) {
        device.beep();
    }
}

Answer 1

将flatMap 替换为concatMap

.concatMap(device -> device.connectForPolicing())

flatMap 使用 merge 运算符。它立即发出所有项目。虽然concatMap 使用concat，但它会按顺序发出项目。 Good article about it.

Answer 2

您可以使用.flatMap(Observable, int) 运算符。

Observable.from(mDevices)
            .flatMap(
                new Func1<Device, Observable<?>>() {
                    @Override
                    public Observable<?> call(Device device) {
                        Log.d(TAG, "connecting for policing");
                        return device.connectForPolicing();
                    }
                },
                1
            )
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Object>() {
                @Override
                public void call(Object o) {
                    Log.d(TAG, "subscribing... ");
                }
            });

int 参数限制了最大并发操作。在这种情况下，它将按顺序处理。

如果您想反复发出蜂鸣声直到设备断开连接，则还需要更改 connectForPolice() 函数：

public Observable<byte[]> connectForPolice(RxBleDevice device) {

    ....

    return device.establishConnection(mContext, false)
            .flatMap(new Func1<RxBleConnection, Observable<byte[]>>() { // once the connection is established ...
                @Override
                public Observable<byte[]> call(RxBleConnection rxBleConnection) {
                    byte[] value = new byte[1];
                    value[0] = (byte) (3 & 0xFF);
                    //Buzz the device
                    return Observable // ... we return an observable ...
                            .defer(new Func0<Observable<byte[]>>() {
                                @Override
                                public Observable<byte[]> call() {
                                    return rxBleConnection.writeCharacteristic(Constants.BUZZER_SELECT, value); // ... (that on each subscription will emit a fresh write characteristic observable) ...
                                }
                            })
                            .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() { // ... which we will subscribe (the Observable.defer()) again ...
                                @Override
                                public Observable<?> call(Observable<? extends Void> observable) {
                                    return observable.delay(10, TimeUnit.SECONDS); // ... after 10 seconds from the previous complete
                                }
                            });
                }
            })
            .onErrorResumeNext(new Func1<Throwable, Observable<? extends byte[]>>() { // if the device will trigger disconnect then a BleDisconnectedException will be thrown ...
                @Override
                public Observable<? extends byte[]> call(Throwable throwable) {
                    return Observable.empty(); // ... in which situation we will just finish the Observable
                }
            });
}