正确使用带有 RxAndroidBLE 的 createNewLongWriteBuilder

Using createNewLongWriteBuilder with RxAndroidBLE Correctly

我有一个设备在虚拟串行端口（使用 BLE）上一次只能接受 20 个字节。如果我没记错的话，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#Radio: QUEUED RxBleRadioOperationCharacteristicLongWrite(107353908) D/RxBle#Radio: STARTED RxBleRadioOperationCharacteristicLongWrite(107353908) I/BLE Controller: Data has been written!

D/RxBle#Radio: FINISHED RxBleRadioOperationCharacteristicLongWrite(107353908)

2017 年 3 月 30 日更新：

我原来的理解是错误的，有很多问题。我发送的是订阅而不是 Observable。

s_noopy指出：

The WriteOperationAckStrategy is effectively an equavilent of ? Observable.repeatWhen() operator. The filtering should happen inside the WOAS to trigger the repeat when ready.`

现状：

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

我的尝试：

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

2017 年 7 月 30 日更新：

修改了 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。看起来好像第一批已成功写入，但即使订阅函数 returns 两次都成功写入字节数组，蓝牙设备也只能看到第一批

日志

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

2017 年 7 月 31 日更新：

已更新以使之前的代码同步 

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)

来自 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： 不使用 Long Write 的替代方法可能如下所示。考虑一个 class ，它将生成需要写入的长 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)