我正在使用 RxAndroidBle 库将值从我的 BLE 设备读取到我的 android 应用程序。应用程序在 BLE 外围设备上进行多次读取，如下面的代码所示。
public void longRead()
{
// if (isConnected())
{
longReadSubscription = connectionObservable
.flatMap(RxBleConnection::discoverServices, (rxBleConnection, services) ->
services.getCharacteristic(UUID_READ_CHARACTERISTIC)
.flatMap(characteristic -> rxBleConnection.queue(new CustomReadOperation(characteristic)))
)
.flatMap(observable -> observable)
.first() // Bug fix for not disconnecting after read.
.observeOn(AndroidSchedulers.mainThread())
.compose(this.bindToLifecycle())
.subscribe(bytes -> {
configureWifiMvpView.showList(bytes);
}, this::onLongReadFailure);
}
}
private static class CustomReadOperation implements RxBleRadioOperationCustom<byte[]> {
private BluetoothGattCharacteristic characteristic;
CustomReadOperation(BluetoothGattCharacteristic characteristic) {
this.characteristic = characteristic;
}
/**
* Reads a characteristic until a special charcter has been found.. This is easily achieve without
* a custom operation. The gain here is that only one operation goes into the RxBleRadio queue
* eliminating the overhead of going on & out of the operation queue.
*/
@NonNull
@Override
public Observable<byte[]> asObservable(BluetoothGatt bluetoothGatt,
RxBleGattCallback rxBleGattCallback,
Scheduler scheduler) throws Throwable {
return readAndObserve(this.characteristic, bluetoothGatt, rxBleGattCallback)
.subscribeOn(scheduler)
.takeFirst(readResponseForMatchingCharacteristic())
.map(byteAssociation -> byteAssociation.second)
.repeatWhen(completed -> completed.delay(200, TimeUnit.MILLISECONDS))
.takeUntil(result -> getSpecialCharacter(result) == true);
}
public boolean getSpecialCharacter(byte[] s) {
boolean result = false;
try {
String str = new String(s, "UTF-8");
if (str == null || str.trim().isEmpty()) {
// System.out.println("Incorrect format of string");
return true;
}
System.out.println(s);
Pattern p = Pattern.compile("]");
Matcher m = p.matcher(str);
result = m.find();
}
}catch(Exception e)
{
System.out.println("exception #####");
}
return result;
}
@NonNull
private Observable<ByteAssociation<UUID>> readAndObserve(BluetoothGattCharacteristic characteristic,
BluetoothGatt bluetoothGatt,
RxBleGattCallback rxBleGattCallback) {
Observable<ByteAssociation<UUID>> onCharacteristicRead = rxBleGattCallback.getOnCharacteristicRead();
return Observable.create(emitter -> {
Subscription subscription = onCharacteristicRead.subscribe(emitter);
emitter.setCancellation(subscription::unsubscribe);
try {
final boolean success = bluetoothGatt.readCharacteristic(characteristic);
if (!success) {
throw new BleGattCannotStartException(bluetoothGatt, BleGattOperationType.CHARACTERISTIC_READ);
}
} catch (Throwable throwable) {
emitter.onError(throwable);
}
}, Emitter.BackpressureMode.BUFFER);
}
private Func1<ByteAssociation<UUID>, Boolean> readResponseForMatchingCharacteristic() {
return uuidByteAssociation -> uuidByteAssociation.first.equals(UUID_READ_CHARACTERISTIC);
}
}
}
现在这一切工作正常，但偶尔如果从 BLE 设备读取的数据超过 600 字节，那么我会在读取完整数据之前调用 Subscribe(bytes->) ，之后我会调用 takeUntil 来检查它字符串末尾的特殊字符并终止，丢失第二块中的所有数据。看起来 RxAndrodiBLE 库维护一个内部缓冲区并继续写入该缓冲区中的所有数据，一旦缓冲区已满，订阅 observable 就会被调用。当读取的数据大小大于缓冲区大小时，这会导致问题。我们如何处理它，这样我们就不会以块的形式获取数据。我们还可以增加缓冲区的大小，以便在调用订阅之前将接收到的所有数据存储在那里。