RxJava中如何根据已有的函数或函数回调创建Observable?

Question

在使用Rxjava过程中，可能已经有很多函数回调，那么怎么根据这些函数回调的参数创建数据流？
比如如果我需要改造onKeyDown(),那么怎么根据传来按键的不同，处理特定用户输入的序列，比如用户输入“1，2，3，4”的时候做特殊处理。

或者有其他的函数回调，怎么将这些函数回调的数据使用bufferDebouncezip等操作符处理数据？

Answer 1

可以这样写

    private BehaviorSubject<Integer> bs;

    private void testSeri() {
        bs = BehaviorSubject.create();
        //每3次 accept 一次
        bs.buffer(3)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(@NonNull List<Integer> ints) throws Exception {
                        StringBuilder sb = new StringBuilder();
                        for (int i = 0; i < ints.size(); i++){
                            sb.append(ints.get(0));
                        }
                        Toast.makeText(TestSubjectActivity.this, sb.toString(), Toast.LENGTH_SHORT).show();
                    }
                });
    }

    @Override
    public boolean onKeyDown(int keyCode, KeyEvent event) {
        bs.onNext(keyCode);
        return super.onKeyDown(keyCode, event);
    }

onKeyDown是Activity的回调，不方便再包装一层，因此用了Subject这种可以【随时随地】发射数据、订阅和发射方便分开写的发射器。对于一般的回调可以这样写，给你个百度定位的回调感受一下

    class LocationObservable implements ObservableOnSubscribe<BDLocation> {
        @Override
        public void subscribe(final ObservableEmitter<BDLocation> e) throws Exception {

            initLocation();
            mLocationClient.registerLocationListener( new BDLocationListener(){
                @Override
                public void onReceiveLocation(BDLocation location) {
                    if (location != null) {
                        mLocationClient.stop();
                        if (!TextUtils.isEmpty(location.getCity())) {
                              e.onNext(location);
                              e.onComplete();
                        }
                    } else {
                        // 定位失败
                        e.onError(new Exception("百度地图 定位失败"));
                    }
                }
            } );
            mLocationClient.start();
        }
    }

对于一般的函数，可以这样

Observable<String> o1 = Observable.fromCallable(new Callable<String>() {
            @Override
            public String call() {
                return func1();
            }
        });
        
        
public String func1(){
    return "ok";
}