DeferredScalarDisposable throws NullPointerException about rxjava HOT 2 CLOSED

Those nulls are internal affairs so unless you can exclude any class with that feature, no idea.

from rxjava.

Sure, thank you. With limited understanding of how Rx2 deals with fusion internally. After give this blog a read - my assumption is, Rx fusion library is adding null in the stream in case of Asynchronous-fusion

Taking below para into consideration

In asynchronous-fusion mode, downstream has to still issue request() calls, but instead of enqueueing the value twice, the value gets generated into the shared queue and the upstream calls onNext() indicating its availability. The value of this call is irrelevant, we use null as a type-neutral value, and can trigger the usual drain() call directly.

I am wondering, if it's safe to assume not to call onNext on downstream observers when Rx fusion code calls onNext(null), getting invoked via upstream observer.

Below is the basic code snippet implementation of otel code in kotlin which deals with fusion.

// downstream  
internal class TracingObserver<T>( 
  private val wrappedObserver: Observer<in T>,
  private val context: Context,
) : BasicFuseableObserver<T, T>(
  wrappedObserver
) {
  override fun onNext(t: T & Any) { // it throws exception when it gets called via fusion
    context.makeCurrent().use { _ -> wrappedObserver.onNext(t) }
  }
......
}

// upstream 
internal class TracingSingleObserver<T>(
  private val actual: SingleObserver<T>,
  private val context: Context,
) : SingleObserver<T>, Disposable {
  private var disposable: Disposable? = null
  override fun onSubscribe(d: Disposable) {
    if (!DisposableHelper.validate(disposable, d)) {
      return
    }
    disposable = d
    actual.onSubscribe(this)
  }

  override fun onSuccess(t: T & Any) {
    context.makeCurrent().use { _ -> actual.onSuccess(t) } // upstream 
  }
....
}

I was wondering, if there are any implication, if i just handle the exception in TracingSingleObserver#onSuccess which will invoke RxFusion with onNext(null) but catches the exception thrown via TracingObserver#onNext.
Can there be any implications if wrappedObserver.onNext(null) isn't called in the RxStream.

Thanks, i need to dig in more!

from rxjava.