rxandroidble - RxAndroidBle long write - 每批写入的回调

Question

我正在做一个长写操作，一次发送 16 个字节的批次。我希望有一个进度条向用户显示长写入的进度，因此每次写入批处理时我都需要某种回调。

从文档来看，它看起来像setWriteOperationAckStrategy这样。但是，在运行以下代码时，我最终只看到一条消息输出到日志中。我在这里做错了什么？

subscription = connection.flatMap(rxBleConnection -> rxBleConnection.createNewLongWriteBuilder()
    .setCharacteristicUuid(uuid)
    .setBytes(bytes)
    .setMaxBatchSize(16)
    .setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
        @Override
        public Observable<Boolean> call(Observable<Boolean> booleanObservable) {
            Log.d("TEST", "batch written");
            return booleanObservable;
        }
    })
    .build()

Answer 1

setWriteOperationAckStrategy 类似于标准 RxJava 的 Observable 转换器。为了保持低分配，我们倾向于修改源 observable，而不是在每批完成时创建一个新的。

subscription = connection.flatMap(rxBleConnection -> rxBleConnection.createNewLongWriteBuilder()
    .setCharacteristicUuid(uuid)
    .setBytes(bytes)
    .setMaxBatchSize(16)
    .setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
        @Override
        public Observable<Boolean> call(Observable<Boolean> booleanObservable) {
            Log.d("TEST", "batch written");
            return booleanObservable
                .doOnNext(new Action1<Boolean>() {
                    @Override
                    public void call(Boolean aBoolean) {
                        Log.d("TEST", "batch written");
                    }
                });
        }
    })
    .build()

Answer 2

我认为这个答案适用于 RxJava，对于 RxJava2，这是更新的 lambda 语法：

.setWriteOperationAckStrategy(booleanObservable -> {
        Log.d("TEST", "batch written");
        return booleanObservable
            .doOnNext(aBoolean -> {
              Log.d("TEST", "batch written");
            });
      }

满的：

.setWriteOperationAckStrategy(new RxBleConnection.WriteOperationAckStrategy() {
  @Override
  public Observable<Boolean> apply(Observable<Boolean> booleanObservable) {
    Log.d("TEST", "batch written");
    return booleanObservable
        .doOnNext(new Consumer<Boolean>() {
           @Override
           public void accept(Boolean aBoolean) {
             Log.d("TEST", "batch written");
           }
         });
  }