What's the best way to debug RxJava problems with not received messages

Question

I've an Android app with multiple observers of type A that subscribe with several Observables of type B. The subscription is done in IO Scheduler and the observation on the Android main thread.

The problem that I've is that randomlly after some work one message emitted by B is never received in A and after some hours of debbuging I can't find the cause.

Relevant code when the problem happens:

"NEXT1" and "NEXT2" are printed but "RECEIVED","ERROR", COMPLETED aren't.

            //The subscription
            B.getMessate()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(A);

            //B
            Observable<msg> getMessage() {
                 return Observable.create(new Observable.OnSubscribe<msg>() {    
                      public void call(Subscriber<? super msg> subscriber) {
                         ...
                         subscriber.onNext(msg)
                         println("NEXT1")
                      }
                 }).doOnNext({ (o) -> println("NEXT2")});
            }



            //A 
            onNext(msg) {
                  //Never called when problem happens
                  println("RECEIVED")
            }
            onError(msg) {
                  //Never called when problem happens
                  println("ERROR")
            }
            onError(msg) {
                  //Never called when problem happens
                  println("COMPLETED")
            }

Anyone has some clue? or any recommendation for debugging that?

What I've checked:

• I've paused the app and checked that all threads to see if one is locked. And aparentlly all working threads are parked and main thread is waiting messages in android message queue.
• Observers never call unsubscribe()

Answer 1

By now I can't reproduce the problem but I've found RxJavaDebug a very good tool for debugging.

It's use is simple: add the library as a dependency and at application start register a listener:

  RxJavaPlugins.getInstance().registerObservableExecutionHook(new DebugHook(new DebugNotificationListener() {
      public Object onNext(DebugNotification n) {
          Log.v(TAG, "onNext on " + n);
          return super.onNext(n);
      }


        public Object start(DebugNotification n) {
            Log.v(TAG, "start on " + n);
            return super.start(n);
        }


        public void complete(Object context) {
            Log.v(TAG, "complete on " + context);
        }

        public void error(Object context, Throwable e) {
            Log.e(TAG, "error on " + context);
        }
  }));

That will log messages while they go between observables and operators.