非同期リスナーをObservable（RxJava2）に変換/返信

Question

Observableオブジェクトに実際のリスナーをラップしたいと思います。 ここでは、テストケースがあります。そのすべてが問題ありません。

private Flowable<Integer> getObservablePhoneState() { 
    return Flowable.create(emitter -> { 

     PhoneStateListener phoneStateListener = new PhoneStateListener() { 
      @Override 
      public void onCallStateChanged(int state, String incomingNumber) { 
       Log.i(TAG, "onCallStateChanged = " + state); 
       emitter.onNext(state); 
      } 
     }; 
     TelephonyManager telephonyManager = (TelephonyManager) getActivity().getSystemService(Context.TELEPHONY_SERVICE); 
     telephonyManager.listen(phoneStateListener, PhoneStateListener.LISTEN_CALL_STATE); 

    }, BackpressureStrategy.BUFFER); 
} 

*** logcat *** 
io.reactivex.exceptions.OnErrorNotImplementedException: 
Attempt to read from field 'android.os.MessageQueue 
android.os.Looper.mQueue' on a null object reference 

Observable.create（）同じエラーで：

@Override 
public void onCreate(@Nullable Bundle savedInstanceState) { 
    getObservablePhoneState() 
     // Run on a background thread 
     .subscribeOn(Schedulers.io()) 
     // Be notified on the main thread 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(integer -> Log.i(TAG, "----- subscribe onNext = " + integer)); 
} 

private Flowable<Integer> getObservablePhoneState() { 
    return Flowable.create(emitter -> { 

     Log.i(TAG, "Emitting 1"); 
     emitter.onNext(1); 

     Log.i(TAG, "Emitting 2"); 
     emitter.onNext(2); 

    }, BackpressureStrategy.BUFFER); 
} 



*** logcat *** 
Emitting 1 
Emitting 2 
----- subscribe onNext = 1 
----- subscribe onNext = 2 

このコードは、エラーが発生します。 おそらく、これはRxJava2 does not support emitting a null valueという事実によるものです。

正しく行うには？

Answer 1

PhoneStateListenerを別のスレッドに作成しないようにするには、subscribeOn(Schedulers.io())を削除する必要があります。これは、フードがmQueueがnullのハンドラを使用してメッセージを送信しようとしているためです。ちょうどお電話

getObservablePhoneState() 
    .subscribe { integer -> Log.i("", "----- subscribe onNext = " + integer) }