search
LoginSignup
66
Help us understand the problem. What are the problem?

More than 5 years have passed since last update.

posted at

updated at

Observable#createは今後使わないほうがいいかもしれない

若干の釣り気味ですが、RxJavaを利用されている方には結構重要そうだよな−と思う話題だったのでQiitaにまとめました。

ことのはじまり

なんだかJakeが過激なこと言ってる気配があったので、背景を調べるために少しissueを漁り始めました。

きっかけのissue
https://github.com/ReactiveX/RxJava/issues/4177

すごく雑に3行でまとめると、

  • 非Rxの世界の非同期処理をラップするのにObservable#create使うよね
  • でもback-puressureまでちゃんと考慮して実装するのクソ難しいよね(僕も多くの失敗から学んでやっと正しい実装をやっと理解したよ)
  • みんなも間違った実装しがちだから、非同期処理をラップしやすいインターフェースあったほうがよくないかな?

という提案です。もう少しだけ文脈を補足すると、前提となっているのはn個のitemをemitするようなObservableについてです。Http通信のやり取りなどrx.Singleで事足りる場合は今回の議論の対象ではなさそうですね。

どうなったの?

2016/07/12地点で最新のver 1.1.7において、Observablue#fromAsyncが追加されました。
インターフェースはObservable#createを使う場合とほとんど一緒ですが、第二引数にBackpressureModeというものを明示的に指定する必要があって、itemを貯めておきたいのか、最新だけ保持したいのかなど決めることが出来ます。ユースケースは様々だと思いますが、BUFFERとLATEST辺りがよく使われそうですね。

        Observable.fromAsync(new Action1<AsyncEmitter<Item>>() {
            @Override
            public void call(final AsyncEmitter<Item> emitter) {
                XyzClient.enqueue(new Callback<List<Item>>() {
                    @Override
                    public void onResponse(Response<Item> response) {
                        emitter.onNext(response.body);
                    }
                    @Override
                    public void onFailure(Call<List<Item>> call, Throwable t) {
                        emitter.onError(t);
                    }
                });

            }
        }, AsyncEmitter.BackpressureMode.BUFFER);

私は普段Promise的な使い方しかほとんど利用しないので、なるほどという気持ちもありつつ、あまりピンときていません。ただ今後Observable#createを敢えて利用するメリットも特になさそうなので、Observablue#fromAsyncで書いていこうという気持ちです。 正直わかってない部分もあるので、より詳しい人が詳細に解説して頂けると嬉しいです! お待ちしています。

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
What you can do with signing up
66
Help us understand the problem. What are the problem?