正确使用createNewLongWriteBuilder与RxAndroidBLE

Question

我有一个设备，一次只能在虚拟串口上接收20个字节(使用BLE)。如果我没有弄错的话，createNewLongWriteBuilder似乎是这方面的完美方法。

以下是我的尝试：

String newNameMsg = "SOME STRING THAT IS LONGER THAN 20 CHARACTERS";

byte[] byteMsg = newNameMsg.getBytes(Charset.forName("UTF-8"));
byte[] endLine = hexStringToByteArray("0D"); // signifies end of line for my device
byte[] newName = new byte[byteMsg.length + endLine.length];
System.arraycopy(byteMsg, 0, newName, 0, byteMsg.length);
System.arraycopy(endLine, 0, newName, byteMsg.length, endLine.length);


connectionObservable
    .flatMap(rxBleConnection -> rxBleConnection.createNewLongWriteBuilder()
    .setCharacteristicUuid(characteristicUuid)
    .setBytes(newName)
    .setMaxBatchSize(20) // my device only accepts 20 characters at a time.
    .setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
            @Override
            public Observable<Boolean> call(Observable<Boolean> booleanObservable) {
                return Observable.just(true); // this is supposed to tell the LongWriteBuilder that we should continue sending data, correct?
            }
        })
    .build()
    )
    .subscribe(
        byteArray -> {
        // Written data.
        Log.i("BLE Controller","Data has been written!");
        },
        throwable -> {
        // Handle an error here.
        }
    );

实际结果:设备不接收任何数据，但日志显示：

D/RxBle#收音机:排队的RxBleRadioOperationCharacteristicLongWrite(107353908) D/RxBle#收音机:已启动RxBleRadioOperationCharacteristicLongWrite(107353908) I/BLE控制器:数据已经写入！ D/RxBle#收音机:已完成的RxBleRadioOperationCharacteristicLongWrite(107353908)

更新资料03/30/2017：

我最初的理解是错误的，而且有很多问题。我发的是订阅而不是可观察的。

s_noopy指出：

WriteOperationAckStrategy实际上是一个等价的？Observable.repeatWhen()算子过滤应该在WOAS内部进行，以便在准备时触发重复。

现状：

我需要等待我的设备清除TX标志，然后才能发送下一批。要做到这一点，我需要实现setWriteOperationAckStrategy，但是在发送下一批之前，我需要读取TX标志以查看是否清楚。

我的尝试：

@Override
public Observable<Boolean> call(Observable<Boolean> objectObservable) {
    return connectionObservable
                .flatMap(rxBleConnection -> rxBleConnection.readCharacteristic(serialTX.getUuid()))
                .observeOn(AndroidSchedulers.mainThread());
}

更新07/30/2017

修改了s_noopy的代码，使我现在有：

final ByteArrayBatchObservable byteArrayBatchObservable = new ByteArrayBatchObservable(newName, 19);
Log.i("BLE Controller","sending updated name"); 
byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
    connectionObservable.flatMap(rxBleConnection -> rxBleConnection.writeCharacteristic(vspT, bytesBatch) 
        .flatMap(writtenBytes -> { // ...and when each batch will be written...
            final Func1<byte[], Boolean> filterFunction = txBytes -> checkIfZero(txBytes);
            return rxBleConnection
                    .readCharacteristic(vspT.getUuid()) 
                    .repeat() // ...and repeat it...
                    .takeUntil(filterFunction)
                    .filter(filterFunction) // ...but don't emit anything until then...
                    .map(readBytes -> writtenBytes); // ...and emit the writtenBytesBatch...
        }
    ), 1)

)
.subscribe(
    byteArray -> {
        // Written data.
        Log.i("BLE Controller","BT has been renamed! :" + byteArray.toString());
    },
    throwable -> {
        // Handle an error here.
        Log.i("BLE Controller","BT rename ERROR");

    }
);

public boolean checkIfZero(byte[] txBytes){
    Log.i("BLE Controller", "checking if tx is cleared: " +txBytes.toString());
    for (byte b : txBytes) {
        if (b != 0) {
            return false;
        }
    }
    return true;
}

现状：

我已经尝试过从使用rxBleConnection转换为使用connectionObservable。似乎第一批已成功写入，但即使订阅函数两次返回成功编写的字节数组，蓝牙设备也只看到第一批。

日志

07-31 13:26:54.434 25060-25060/com.packet.sniffer I/BLE Controller: sending updated name
07-31 13:26:54.434 264-467/? I/ThermalEngine: Sensor:pa_therm1:33000 mC
07-31 13:26:54.440 25060-25060/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicWrite(99278425)
07-31 13:26:54.443 25060-25187/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicWrite(99278425)
07-31 13:26:54.458 25060-25060/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicWrite(56755989)
07-31 13:26:54.611 25060-25072/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicWrite characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 13:26:54.614 25060-25243/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicRead(32706505)
07-31 13:26:54.614 25060-25187/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicWrite(99278425)
07-31 13:26:54.615 25060-25187/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicWrite(56755989)
07-31 13:26:54.714 25060-25071/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicWrite characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 13:26:54.721 25060-25243/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicRead(39309874)
07-31 13:26:54.723 25060-25187/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicWrite(56755989)
07-31 13:26:54.724 25060-25187/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicRead(32706505)
07-31 13:26:54.809 1980-2244/com.android.bluetooth I/bt_btif_gatt: set_read_value unformat.len = 20 
07-31 13:26:54.811 25060-25072/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicRead characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 13:26:54.813 25060-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@c805018
07-31 13:26:54.813 25060-25243/com.packet.sniffer I/BLE Controller: BT has been renamed! :[B@5999a71
07-31 13:26:54.813 25060-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@c805018
07-31 13:26:54.816 25060-25187/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicRead(32706505)
07-31 13:26:54.817 25060-25187/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicRead(39309874)
07-31 13:26:54.908 1980-2244/com.android.bluetooth I/bt_btif_gatt: set_read_value unformat.len = 20 
07-31 13:26:54.910 25060-25071/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicRead characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 13:26:54.911 25060-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@a1f9bcf
07-31 13:26:54.911 25060-25243/com.packet.sniffer I/BLE Controller: BT has been renamed! :[B@208995c
07-31 13:26:54.911 25060-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@a1f9bcf
07-31 13:26:54.914 25060-25187/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicRead(39309874)
07-31 13:26:56.467 1980-2388/com.android.bluetooth D/HeadsetStateMachine: Disconnected process message: 10, size: 0

更新07/31/2017：

更新以使以前的代码同步。

byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
    connectionObservable.flatMap(rxBleConnection -> rxBleConnection.writeCharacteristic(vspT, bytesBatch)
                            .flatMap(writtenBytes -> { // ...and when each batch will be written...
                                    final Func1<byte[], Boolean> filterFunction = txBytes -> checkIfZero(txBytes);
                                    return rxBleConnection
                                            .readCharacteristic(vspT.getUuid())
                                            .repeat() // ...and repeat it...
                                            .takeUntil(filterFunction)
                                            .filter(filterFunction) // ...but don't emit anything until then...
                                            .map(readBytes -> writtenBytes); // ...and emit the writtenBytesBatch...
                                }
                            ))
, 1)

不幸的是，仍然无法工作，下面是相关的日志：

4:55:35.108 25084-25084/com.packet.sniffer I/BLE Controller: sending updated name
07-31 14:55:35.110 25084-25084/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicWrite(135224277)
07-31 14:55:35.113 25084-25209/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicWrite(135224277)
07-31 14:55:35.113 25084-25084/com.packet.sniffer I/BluetoothLEController: verify connectivity
07-31 14:55:35.219 25084-25096/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicWrite characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 14:55:35.225 25084-25243/com.packet.sniffer D/RxBle#Radio:   QUEUED RxBleRadioOperationCharacteristicRead(155469961)
07-31 14:55:35.228 25084-25209/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicWrite(135224277)
07-31 14:55:35.229 25084-25209/com.packet.sniffer D/RxBle#Radio:  STARTED RxBleRadioOperationCharacteristicRead(155469961)
07-31 14:55:35.316 25084-25097/com.packet.sniffer D/RxBle#BluetoothGatt: onCharacteristicRead characteristic=569a2000-b87f-490c-92cb-11ba5ea5167c status=0
07-31 14:55:35.317 25084-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@f0396a7
07-31 14:55:35.317 25084-25243/com.packet.sniffer I/BLE Controller: BT has been renamed! :[B@8983754
07-31 14:55:35.317 25084-25243/com.packet.sniffer I/BLE Controller: checking if tx is cleared: [B@f0396a7
07-31 14:55:35.320 25084-25209/com.packet.sniffer D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicRead(155469961)

Answer 1

来自setWriteOperationAckStrategy()的文档

If you want to delay the next batch use provided observable and add some custom behavior (delay, waiting for a message from the device, etc.)

从你的片段来看：

.setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
    @Override
    public Observable<Boolean> call(Observable<Boolean> booleanObservable) {
        return Observable.just(true); // this is supposed to tell the LongWriteBuilder that we should continue sending data, correct?
    }
})

您对下一批货的延误不感兴趣。在这种情况下，您可以忽略将WriteOperationAckStrategy设置为文档状态：If this is not specified - the next batch of bytes is written right after the previous one has finished.，它等于：

.setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
    @Override
    public Observable<Boolean> call(Observable<Boolean> objectObservable) {
        return objectObservable;
    }
})

编辑:要延迟下一批写入，需要延迟ACK信号。由于长写操作确保在写操作之间不会发生其他操作，唯一可能的选择是对特定的通知/指示或其他旁通道事件进行中继。

编辑1:不使用长写的替代方法可能如下所示。考虑一个类，该类将生成需要编写的长byte[]的批处理：

public class ByteArrayBatchObservable extends Observable<byte[]> {

    public ByteArrayBatchObservable(@NonNull final byte[] bytes, final int maxBatchSize) {
        super(SyncOnSubscribe.createSingleState(
                new Func0<ByteBuffer>() {
                    @Override
                    public ByteBuffer call() {
                        return ByteBuffer.wrap(bytes);
                    }
                },
                new Action2<ByteBuffer, Observer<? super byte[]>>() {
                    @Override
                    public void call(ByteBuffer byteBuffer, Observer<? super byte[]> observer) {
                        int nextBatchSize = Math.min(byteBuffer.remaining(), maxBatchSize);
                        if (nextBatchSize == 0) {
                            observer.onCompleted();
                            return;
                        }
                        final byte[] nextBatch = new byte[nextBatchSize];
                        byteBuffer.get(nextBatch);
                        observer.onNext(nextBatch);
                    }
                }
        ));
    }
}

然后，在代码中可以使用类似的代码来：

final ByteArrayBatchObservable byteArrayBatchObservable = new ByteArrayBatchObservable(longByteArrayToWrite, maxBatchSize); // create an observable that will make chunks of data small enough to write at once
return byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
                rxBleConnection.writeCharacteristic(characteristicUuid, bytesBatch) // ...write them on characteristic...
                        .flatMap(writtenBytes -> { // ...and when each batch will be written...
                                    final Func1<byte[], Boolean> filterFunction = txBytes -> txBytes.length == 1 && txBytes[0] == 0;
                                    return rxBleConnection
                                            .readCharacteristic(txCharacteristicUuid) // ...start reading the TX characteristic...
                                            .repeat() // ...and repeat it...
                                            .takeUntil(filterFunction) // ...until the read value will indicate that the device is ready for the next batch...
                                            .filter(filterFunction) // ...but don't emit anything until then...
                                            .map(readBytes -> writtenBytes); // ...and emit the writtenBytesBatch...
                                }
                        ),
        1 // ...to be sure that only one .writeCharacteristic() will be subscribed at any given time
);

编辑2:根据您在写长号时想要达到的目标：

1. 写一批数据
2. 执行设备就绪标志的读取。
3. 如果设备尚未准备好，请返回到2。
4. 如果还有更多的数据需要写入，请返回到1。

您可以在日志中看到的内容：

1. 成批数据写入
2. 另一批数据被写入。
3. 读取执行
4. 执行另一个读取。

这正是您在代码中所拥有的，因为您没有同步单个批。这可能是由于放置.flatMap(Observable, 1)的位置造成的复制粘贴错误。

final ByteArrayBatchObservable byteArrayBatchObservable = new ByteArrayBatchObservable(newName, 19);
Log.i("BLE Controller","sending updated name"); 
byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
    connectionObservable.flatMap(rxBleConnection -> rxBleConnection.writeCharacteristic(vspT, bytesBatch) 
        .flatMap(writtenBytes -> { // ...and when each batch will be written...
            final Func1<byte[], Boolean> filterFunction = txBytes -> checkIfZero(txBytes);
            return rxBleConnection
                    .readCharacteristic(vspT.getUuid()) 
                    .repeat() // ...and repeat it...
                    .takeUntil(filterFunction)
                    .filter(filterFunction) // ...but don't emit anything until then...
                    .map(readBytes -> writtenBytes); // ...and emit the writtenBytesBatch...
        }
    ), 1)

)

在这里，为了实现同步行为，代码应该只是有点不同：

final ByteArrayBatchObservable byteArrayBatchObservable = new ByteArrayBatchObservable(newName, 19);
Log.i("BLE Controller","sending updated name"); 
byteArrayBatchObservable.flatMap(bytesBatch -> // using batches of data to write...
    connectionObservable.flatMap(rxBleConnection -> rxBleConnection.writeCharacteristic(vspT, bytesBatch) 
        .flatMap(writtenBytes -> { // ...and when each batch will be written...
            final Func1<byte[], Boolean> filterFunction = txBytes -> checkIfZero(txBytes);
            return rxBleConnection
                    .readCharacteristic(vspT.getUuid()) 
                    .repeat() // ...and repeat it...
                    .takeUntil(filterFunction)
                    .filter(filterFunction) // ...but don't emit anything until then...
                    .map(readBytes -> writtenBytes); // ...and emit the writtenBytesBatch...
        }
    ))
    .take(1)
, 1)