将异步侦听器转换/包装为 Observable (RxJava2)

Question

我想为 Observable 对象包装一个真正的监听器。对于新手这里是一个测试用例，有他一切都很好。

@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

此代码产生错误：

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()同样的错误。也许这是由于 RxJava2 does not support emitting a null value.

怎么做才对？

Answer 1

您应该删除 subscribeOn(Schedulers.io()) 以避免在不同的线程中创建 PhoneStateListener，因为在后台尝试使用 mQueue 为空的处理程序发送消息。直接打电话

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

将异步侦听器转换/包装为 Observable (RxJava2)

Convert/ wrap async listener to Observable (RxJava2)

android

listeners

rx-java2