MissingBackpressureException on version 0.5

[mangolas]

I updated our application to use UX beta 0.5.1 from 0.4 and have started occasionally getting exception below which crashes whole app (now while developing on Smart Controller without connecting to drone).

This didn't happen on 0.4. Any idea how to cure the situation, widget backpressure should not kill the whole app.

dji.thirdparty.io.reactivex.exceptions.MissingBackpressureException: Could not deliver value due to lack of requests
        at dji.thirdparty.io.reactivex.processors.BehaviorProcessor$BehaviorSubscription.test()
        at dji.thirdparty.io.reactivex.processors.BehaviorProcessor$BehaviorSubscription.emitNext()
        at dji.thirdparty.io.reactivex.processors.BehaviorProcessor.onNext()
        at dji.ux.beta.core.util.DataProcessor.onNext(DataProcessor.java:64)
        at dji.ux.beta.core.widget.compass.CompassWidgetModel.updateStates(CompassWidgetModel.kt:188)
        at dji.ux.beta.core.widget.compass.CompassWidgetModel.onSensorChanged(CompassWidgetModel.kt:240)
        at android.hardware.SystemSensorManager$SensorEventQueue.dispatchSensorEvent(SystemSensorManager.java:699)
        at android.os.MessageQueue.nativePollOnce(Native Method)
        at android.os.MessageQueue.next(MessageQueue.java:323)
        at android.os.Looper.loop(Looper.java:136)
        at android.app.ActivityThread.main(ActivityThread.java:6240)
        at java.lang.reflect.Method.invoke(Native Method)
        at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1000)
        at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:868)

[mangolas]

Ok, after bit of digging, the onSensorChanged in CompassWidgetModel is called a lot, after moving `updateStates()' in if block, haven't got this exception for a while, so not sure if this a real fix but seems to help.

   override fun onSensorChanged(event: SensorEvent) {
        // Update the mobile device azimuth when updated by the sensor
        var sensorValue = latestSensorValue
        if (event.sensor.type == Sensor.TYPE_ORIENTATION) {
            sensorValue = event.values[0]
        } else if (event.sensor.type == Sensor.TYPE_ROTATION_VECTOR) {
            MathUtil.getRotationMatrixFromVector(rotations, event.values)
            SensorManager.getOrientation(rotations, values)
            sensorValue = Math.toDegrees(values[0].toDouble()).toFloat()
        }
        if (abs(sensorValue - latestSensorValue) > SENSOR_SENSITIVE_PARAM) {
            latestSensorValue = sensorValue
            val rotation: Int = getDisplayRotation()
            if (rotation == Surface.ROTATION_270) {
                sensorValue += HALF_TURN.toFloat()
            }
            if (DJIDeviceUtil.isSmartController()) {
                sensorValue += QUARTER_TURN.toFloat()
            }
            val mobileDeviceAzimuth = sensorValue + QUARTER_TURN
            mobileDeviceAzimuthProcessor.onNext(mobileDeviceAzimuth)
            updateStates()
        }
    }

[Dommerjon]

I see something similar, and will try out your suggestion

dji.thirdparty.io.reactivex.exceptions.MissingBackpressureException: Could not deliver value due to lack of requests at dji.thirdparty.io.reactivex.processors.BehaviorProcessor$BehaviorSubscription.test() at dji.thirdparty.io.reactivex.processors.BehaviorProcessor$BehaviorSubscription.emitNext() at dji.thirdparty.io.reactivex.processors.BehaviorProcessor.onNext() at dji.ux.beta.core.util.DataProcessor.onNext(DataProcessor.java:64) at dji.ux.beta.core.widget.compass.CompassWidgetModel.updateStates(CompassWidgetModel.kt:188) at dji.ux.beta.core.base.WidgetModel.lambda$registerKey$3$WidgetModel(WidgetModel.java:263) at dji.ux.beta.core.base.-$$Lambda$WidgetModel$XEPTgS4qGvp2HcVVvkA94USB8hI.accept(lambda) at dji.thirdparty.io.reactivex.internal.subscribers.LambdaSubscriber.onNext() at dji.thirdparty.io.reactivex.internal.operators.flowable.FlowableDoOnEach$DoOnEachSubscriber.onNext() at dji.thirdparty.io.reactivex.internal.operators.flowable.FlowableDoOnEach$DoOnEachSubscriber.onNext() at dji.thirdparty.io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onNext() at dji.thirdparty.io.reactivex.internal.operators.flowable.FlowableCreate$LatestAsyncEmitter.drain() at dji.thirdparty.io.reactivex.internal.operators.flowable.FlowableCreate$LatestAsyncEmitter.onNext() at dji.ux.beta.core.base.DJISDKModel.lambda$registerKey$4(DJISDKModel.java:299) at dji.ux.beta.core.base.-$$Lambda$DJISDKModel$WU8jUzG-gcee5Gj22M7dXbNUzZM.onValueChange(lambda) at dji.internal.dgh.dfh.run() at android.os.Handler.handleCallback(Handler.java:755) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:167) at android.os.HandlerThread.run(HandlerThread.java:61)

[Dommerjon]

I implemented your "work around", and I havn't seen the exception since Thanks !

[mangolas]

Good that it helped, hope this get's addressed in actual codebase.

[jeryini]

Hello @mangolas ! I've seen the same issue in our app in production, I also recommend using onBackpressureDrop/onBackpressureLatest (rxJava official docs) when subscribing to widget events via getWidgetStateUpdate. Though haven't been able to test if using this actually solves the problem as I can't repeat the issue locally.

[jeryini]

I still get this issue sometimes, even with the fix. I think the answer lies in the following (see official rxjava docs):

Note that this processor signals MissingBackpressureException if a particular Subscriber is not ready to receive onNext events. To avoid this exception being signaled, use offer(Object) to only try to emit an item when all Subscribers have requested item(s).

Backpressure:
The BehaviorProcessor does not coordinate requests of its downstream Subscribers and expects each individual Subscriber is ready to receive onNext items when onNext(Object) is called. If a Subscriber is not ready, a MissingBackpressureException is signalled to it. To avoid overflowing the current Subscribers, the conditional offer(Object) method is available that returns true if any of the Subscribers is not ready to receive onNext events. If there are no Subscribers to the processor, offer() always succeeds. If the BehaviorProcessor is (optionally) subscribed to another Publisher, this upstream Publisher is consumed in an unbounded fashion (requesting Long.MAX_VALUE).

Therefore I recommend calling offer instead of onNext inside DataProcessor.onNext method.

[ccjimmy777]

I have also encountered this exception these days. So is there any reliable solution now?

[Dommerjon]

I have also encountered this exception these days. So is there any reliable solution now?

I followed the suggestion made by mangolas commented on 8 Mar 2021

to move the updateStates() in the onSensorChanged to the if block, and this has worked well for me 👍