MissingBackpressureException甚至.onBackpressureDrop（）被添加？

Question

我有MissingBackpressureException的問題。

我添加了幾個.onBackpressureDrop（）只是爲了測試，但仍然得到例外。

我添加了RxJavaHooks.enableAssemblyTracking（）以獲取更多日誌細節。

例外是1-3分鐘後投擲。

任何想法這段代碼有什麼問題？

Thx尋求幫助。

代碼，其中異常被扔：

Subscription notifySubscription = connection.setupNotification(notifyCharacteristic) 
            .onBackpressureBuffer() 
            .doOnNext(new Action1<Observable<byte[]>>() { 
             @Override 
             public void call(Observable<byte[]> observable) { 
              Log.d(TAG, "Notifications set, calling bypassConnect()"); 

              // Bypass connect in WB to make it aware of this new device 
              String wbAddress = addressMap.getOrCreateWbAddress(bleMac); 
              bleWrapper.bypassConnect(wbAddress); 
             } 
            }) 
            .onBackpressureDrop() 
            .flatMap(new Func1<Observable<byte[]>, Observable<byte[]>>() { 
             @Override 
             public Observable<byte[]> call(Observable<byte[]> observable) { 
              return observable; 
             } 
            }) 
            .onBackpressureDrop() 
            .subscribe(new Action1<byte[]>() { 
             @Override 
             public void call(byte[] bytes) { 
              dataAvailable(bleMac, bytes); 
             } 
            }, new Action1<Throwable>() { 
             @Override 
             public void call(Throwable throwable) { 
              Log.e(TAG, "dataAvailable() Error: ", throwable); 
             } 
            }); 

登錄與：

rx.exceptions.MissingBackpressureException 
at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:325) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.queueScalar(OperatorMerge.java:379) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:361) 
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846) 
at com.jakewharton.rxrelay.RelaySubscriptionManager$RelayObserver.onNext(RelaySubscriptionManager.java:205) 
at com.jakewharton.rxrelay.PublishRelay.call(PublishRelay.java:47) 
at com.jakewharton.rxrelay.SerializedAction1.call(SerializedAction1.java:84) 
at com.jakewharton.rxrelay.SerializedRelay.call(SerializedRelay.java:20) 
at com.polidea.rxandroidble.internal.connection.RxBleGattCallback$4.onCharacteristicChanged(RxBleGattCallback.java:139) 
at android.bluetooth.BluetoothGatt$1.onNotify(BluetoothGatt.java:438) 
at android.bluetooth.IBluetoothGattCallback$Stub.onTransact(IBluetoothGattCallback.java:399) 
at android.os.Binder.execTransact(Binder.java:453) 
Caused by: rx.exceptions.AssemblyStackTraceException: Assembly trace: 
at rx.Observable.unsafeCreate(Observable.java:162) 
at rx.Observable.lift(Observable.java:299) 
at rx.Observable.merge(Observable.java:2572) 
at rx.Observable.merge(Observable.java:2914) 
at rx.Observable.merge(Observable.java:2637) 
at com.polidea.rxandroidble.internal.connection.RxBleGattCallback.getOnCharacteristicChanged(RxBleGattCallback.java:311) 
at com.polidea.rxandroidble.internal.connection.NotificationAndIndicationManager.observeOnCharacteristicChangeCallbacks(NotificationAndIndicationManager.java:186) 
at com.polidea.rxandroidble.internal.connection.NotificationAndIndicationManager.access$400(NotificationAndIndicationManager.java:31) 
at com.polidea.rxandroidble.internal.connection.NotificationAndIndicationManager$1$1.call(NotificationAndIndicationManager.java:110) 
at com.polidea.rxandroidble.internal.connection.NotificationAndIndicationManager$1$1.call(NotificationAndIndicationManager.java:107) 
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:69) 
at rx.observers.Subscribers$5.onNext(Subscribers.java:235) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355) 
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846) 
at rx.observers.SerializedObserver.onNext(SerializedObserver.java:91) 
at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:94) 
at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.innerNext(OnSubscribeConcatMap.java:182) 
at rx.internal.operators.OnSubscribeConcatMap$ConcatMapInnerScalarProducer.request(OnSubscribeConcatMap.java:366) 
at rx.internal.producers.ProducerArbiter.setProducer(ProducerArbiter.java:126) 
at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.drain(OnSubscribeConcatMap.java:278) 
at rx.internal.operators.OnSubscribeConcatMap$ConcatMapSubscriber.onNext(OnSubscribeConcatMap.java:144) 
at rx.internal.operators.OnSubscribeFromArray$FromArrayProducer.slowPath(OnSubscribeFromArray.java:100) 
at rx.internal.operators.OnSubscribeFromArray$FromArrayProducer.request(OnSubscribeFromArray.java:63) 
at rx.Subscriber.setProducer(Subscriber.java:211) 
at rx.Subscriber.setProducer(Subscriber.java:205) 
at rx.internal.operators.OnSubscribeFromArray.call(OnSubscribeFromArray.java:32) 
at rx.internal.operators.OnSubscribeFromArray.call(OnSubscribeFromArray.java:24) 
at rx.Observable.unsafeSubscribe(Observable.java:10256) 
at rx.internal.operators.OnSubscribeConcatMap.call(OnSubscribeConcatMap.java:94) 
at rx.internal.operators.OnSubscribeConcatMap.call(OnSubscribeConcatMap.java:42) 
at rx.Observable.unsafeSubscribe(Observable.java:10256) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:248) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:148) 
at rx.in 

Answer 1

看起來像由rxandroidble返回不支持背壓Observable但flatMap期望它，所以你必須在flatMap內部應用.onBackpressureDrop：

.flatMap(observable -> observable.onBackpressureDrop()) 

請注意，在之外應用運營商flatMap通常不會影響合併源時發生的情況。