3.x: SwitchMapInnerObserver.queue.offer NullPointerException #7596
Description
RxJava 3.1.0
I'm getting this NPE:
java.lang.NullPointerException: Attempt to invoke interface method 'boolean SimpleQueue.offer(java.lang.Object)' on a null object reference
at io.reactivex.rxjava3.internal.operators.observable.ObservableSwitchMap$SwitchMapInnerObserver.onNext(ObservableSwitchMap.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableMap$MapObserver.onNext(ObservableMap.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableScanSeed$ScanSeedObserver.onSubscribe(ObservableScanSeed.java)
at io.reactivex.rxjava3.subjects.PublishSubject.v1(PublishSubject.java:6)
at io.reactivex.rxjava3.core.Observable.subscribe(Observable.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableScanSeed.subscribeActual(ObservableScanSeed.java)
at io.reactivex.rxjava3.core.Observable.subscribe(Observable.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableDefer.subscribeActual(ObservableDefer.java)
at io.reactivex.rxjava3.core.Observable.subscribe(Observable.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableMap.subscribeActual(ObservableMap.java)
at io.reactivex.rxjava3.core.Observable.subscribe(Observable.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableSwitchMap$SwitchMapObserver.onNext(ObservableSwitchMap.java)
at io.reactivex.rxjava3.observers.SerializedObserver.onNext(SerializedObserver.java)
at io.reactivex.rxjava3.internal.operators.observable.ObservableThrottleFirstTimed$DebounceTimedObserver.onNext(ObservableThrottleFirstTimed.java)
at io.reactivex.rxjava3.internal.util.NotificationLite.accept(NotificationLite.java)
at io.reactivex.rxjava3.subjects.BehaviorSubject$BehaviorDisposable.test(BehaviorSubject.java)
at io.reactivex.rxjava3.subjects.BehaviorSubject$BehaviorDisposable.emitNext(BehaviorSubject.java)
at io.reactivex.rxjava3.subjects.BehaviorSubject.onNext(BehaviorSubject.java)
at onAndroidWidgetUpdateByOs(Main.java)
This is the abstracted code that causes it:
private val refreshModelsSubject = BehaviorSubject.createDefault(Unit)
private val refreshObservable = refreshModelsSubject.throttleFirst(THROTTLE_REFRESH_S, TimeUnit.MILLISECONDS)
private val dimensionsChangedForWidgetIdSubject = PublishSubject.create<Pair<Int, Bundle>>()
fun onAndroidWidgetUpdateByOs() {
refreshModelsSubject.onNext(Unit)
}
// ... in the initialization code:
fun init() {
refreshObservable.switchMap {
Observable.defer {
dimensionsChangedForWidgetIdSubject
.scan(initialDimensionsByWidgetId()) { previousDimensionsByWidgetId, pair ->
// ... scan code
}
.map {
// ... map code
}
}
}
}
This issue happens very rarely and I can't repro it locally.
I have been reading the Observables involved in the crash for a few hours and I think what happens is that:
refreshModelsSubjectemits. This causes a chain of events that will lead toObservableScanSeed.onSubscribecode being called with theSwitchMapInnerObserverbeing passed to it (indirectly insideObservableMap).- While inside
ObservableScanSeed.onSubscribe, something will dispose of theSwitchMapInnerObserverand then the call toSwitchMapInnerObserver.onSubscribethat happens fromObservableScanSeed.onSubscribewill not create thequeueobject inSwitchMapInnerObserver. ObservableScanSeed.onSubscribewill callSwitchMapInnerObserver.onNextand crash happens
Is my assessment correct? If so, what kind of mistakes I'm making here and how do I mitigate them?
I also saw this issue, which is similar, but seems like it was caused by a non-standard operator (which I'm not using).