ネストされたインタフェースコールバックをRxJavaで置き換える

RxJava置換ネストインタフェースコールバック
プロジェクトで2回または3回以上インタフェースにアクセスする必要がある場合があります.この場合、同期方式でインタフェースを移動する場合は、ネストされたインタフェースコールバックを実現する必要があります.
従来のコードは次のように実装されています.

コールバックインタフェース

を定義する.

    public interface Callback {
        void onSuccess(Object result);

        void onFail(Exception e);
    }

パッケージの2つのインタフェースへのアクセス

    //       
    public void firstApi(Callback callback) {
        Request request = new Request.Builder().url("http://www.example1.com").build();
        new OkHttpClient().newCall(request).enqueue(new okhttp3.Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                callback.onFail(e);
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                if (response.isSuccessful()) {
                    byte[] bytes = response.body().bytes();
                    if (bytes != null) {
                        callback.onSuccess(bytes);
                    }
                } else {
                    callback.onFail(new IllegalStateException("response code is not 200"));
                }
            }
        });
    }

    //       
    public void secondApi(Callback callback) {
        Request request = new Request.Builder().url("http://www.example2.com").build();
        new OkHttpClient().newCall(request).enqueue(new okhttp3.Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                callback.onFail(e);
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                if (response.isSuccessful()) {
                    String jsonString = response.body().string();
                    callback.onSuccess(jsonString);
                } else {
                    callback.onFail(new IllegalStateException("response code is not 200"));
                }
            }
        });
    }

で使用すると、ネストされたインタフェースが

にコールバックする.

    public void test() {
        firstApi(new Callback() {
            @Override
            public void onSuccess(Object result) {
                secondApi(new Callback() {
                    @Override
                    public void onSuccess(Object result) {
                        //do something...
                    }

                    @Override
                    public void onFail(Exception e) {

                    }
                });
            }

            @Override
            public void onFail(Exception e) {

            }
        });
    }

このようなコールバックは2つならまだしも、3つ以上は見られない.
RxJava方式に変更

は主にflatMap切替Observableを利用して被観察テーマ

を置き換える.

ネストするたびにflatMap

が呼び出される

/**
 * @author zjy
 * @date 2018/6/25
 */
public class RxJavaTestActivity extends AppCompatActivity {

    private TextView mTvInfo;

    public static void start(Context context) {
        Intent intent = new Intent(context, RxJavaTestActivity.class);
        context.startActivity(intent);
    }

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_test);
        mTvInfo = findViewById(R.id.tv_info);
    }

    @SuppressLint("CheckResult")
    public void onTest(View view) {
        firstFun()
                .flatMap((Function>) this::secondFun)//    ：  flatMap          （       Observable）
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(s -> mTvInfo.append("accept result: " + s));
    }

    //       
    public Observable firstFun() {
        return Observable.create(emitter -> new Thread(() -> {
            //        
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            final String result = "first observable class
";
            runOnUiThread(() -> mTvInfo.append("first result: " + result));
            emitter.onNext(result);
            emitter.onComplete();
        }).start());
    }

    //       
    public Observable secondFun(String param) {
        return Observable.create((ObservableEmitter emitter) -> new Thread(() -> {
            //        
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            final String result = "second observable class
";
            runOnUiThread(() -> {
                mTvInfo.append("second: upstream param \"" + param + "\"");
                mTvInfo.append("second result: " + result);
            });
            emitter.onNext(result);
            emitter.onComplete();
        }).start());
    }

}