2016/6/9 タイトル変更しました

前回の記事でRxJavaを使い始めましたが、初めてのRxということも有りなかなか理解するのに時間がかかりました。

そこで自分がわかりやすいようにまとめてみようと思います。

RxやRxJavaって何？って人はここを見てください。

私は、非同期処理(HTTPリクエストやDB問い合わせなど)や複数の値の処理を短く書ける考え方/ライブラリとして捉えています。

Rxで考えるにあたって一番重要だと思うこと

RxJava(Rx)でコードを書く上で一番理解する妨げになったのは、流れているデータがどういった物(型)なのかわかりにくい。ということです。

逆に言えば、流れているデータさえ把握してしまえばコードを書くのは結構楽になるということです。

基本例

例えば以下の様なコード。

Observable.from(new String[]{"hoge", "fuga", "test", "sample"})
.map(new Func1<String, Integer>(){
    @Override
    public Integer call(String s){
        return s.length();
    }
}).subscribe(new Action1<Integer>(){
    @Override
    public void call(Integer i){
        System.out.println(i);
    }
});

RxJavaなのにラムダ式じゃねーのかよという意見もあるかとは思いますが、この記事ではラムダ式でコードは書きません。

ラムダ式で書かれた記事を見て自分が余計に混乱したので。

さて、上のコードを見て自分がまず思ったのは、「どこでイレテートみたいなのをしてるの？」ということです。

.map()内で配列のString要素を受け取りIntegerを返しているということはなんとなくですがわかります。

じゃあどこで配列内の各要素を取り出して渡しているのかがわかりませんでした。

ですが、これは比較的すぐ理解できました。

上のコードを少し変えてみます

Observable<String> mObservable = Observable.from(new String[]{"hoge", "fuga", "test", "sample"});
mObservable.map(new Func1<String, Integer>(){
    @Override
    public Integer call(String s){
        return s.length();
    }
}).subscribe(new Action1<Integer>(){
    @Override
    public void call(Integer i){
        System.out.println(i);
    }
});

一行目で、

Observable<String> mObservable = Observable.from(new String[]{"hoge", "fuga", "test", "sample"})

となっており、Observable.fromが何か内部的にやってくれて、イレテート的なことをやってくれるObservable型に変換するという解釈です。

そして、"hoge" "fuga" "test" "sample"がそれぞれ.map()で文字列から文字列の長さへと置き換えられ、.subscribe()でprintlnされています。

イメージとしては工場のラインでしょうか。

Observable.from(new String[]{"hoge", "fuga", "test", "sample"})

で、入荷した

String[]{"hoge", "fuga", "test", "sample"}

という、「文字列の配列」という「材料」の、「配列」という「梱包」を解きます。

実際にラインに流すのは"hoge" "fuga" "test" "sample"という加工対象。

そして.map()内のcall()で加工されて次のオペレータへ渡される....という解釈をしています。

では、以下のような二重配列ではどうでしょうか。

Observable.from(new String[][]{{"hoge", "fuga"}, {"test", "sample"}})

Observable.fromで解いてくれる「梱包」は一番外側の「梱包」だけです。

つまり、.map()に流れてくるのは{"hoge", "fuga"}　{"test", "sample"}という2つの配列です。

この場合、この配列を加工して出せる値（要素数とか）なら問題ないのですが、上のように、欲しいのが各文字列に関する何かだった場合、それを得ることは出来ません。

ではどうすれば良いのか。

答えは単純、もう一度配列という梱包を解いてあげれば良いのです。

そこで登場するのが.flatMap()というオペレータ。

どうなるのか実際にコードで見てみましょう。

Observable<String[]> mObservable = Observable.from(new String[][]{{"hoge", "fuga"}, {"test", "sample"}});
mObservable.flatMap(new Func1<String[], Observable<String>>(){
    @Override
    public Observable<String> call(String[] s){
        return Observable.from(s);
    }
}).map(new Func1<String, Integer>() {
    @Override
    public Integer call(String s){
　　　　return s.length();
    }
}).subscribe(new Action1<Integer>(){
    @Override
    public void call(Integer i){
        System.out.println(i);
    }
});

変更点は、

.map(new Func1<String, Integer>() {
    @Override
    public Integer call(String s){
　　　　return s.length();
    }
}).subscribe(new Action1<Integer>(){
    @Override
    public void call(Integer i){
        System.out.println(i);
    }
});

の前に

.flatMap(new Func1<String[], Observable<String>>(){
    @Override
    public Observable<String> call(String[] s){
        return Observable.from(s);
    }
})

が追加されていることです。

.flatMap()オペレータとは、Observableを新たに生成して返せる。つまり、梱包を解いて、解いた物を次のオペレータへ流してくれるオペレータです。
(※この解釈が正しいのかどうか自分でもイマイチよくわかっていません。この意味不明な説明よりコード見たほうが直感的だと思います。)

最後に.subscribe()オペレータについて解説します。

このオペレータは、.map()等のオペレータを通し、加工が終わって流れてきた「材料」の最終処理を行うオペレータです。

最終処理と書いているのは、.subscribe()以降材料が流れることがないからです。このオペレータ内で材料をprintlnで表示したり、必要としているオブジェクトへ値を渡したりします。

で、ここは自分も理解できていない部分ですが、.subscribe()の引数には2種類のクラスを与えることができます。

ひとつは、上のコードで登場しているAction1<T>

もうひとつは、Observer<T>です。

Observable<String[]> mObservable = Observable.from(new String[][]{{"hoge", "fuga"}, {"test", "sample"}});
mObservable.flatMap(new Func1<String[], Observable<String>>(){
    @Override
    public Observable<String> call(String[] s){
        return Observable.from(s);
    }
}).map(new Func1<String, Integer>() {
    @Override
    public Integer call(String s){
        return s.length();
    }
}).subscribe(new Observer<Integer>() {
    @Override
    public void onCompleted() {
        
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer integer) {

    }
});

Action1<T>と比べると、実装メソッドがまったく違いますね。

ここの使い分けがどうすればいいのかよくわかっていません。

Observer<T>の場合、onErrorやonCompletedなどのメソッドがあるため、データソースがAPIやDBから取得するもの（=データ取得に失敗する可能性のあるもの）やデータの加工が終わったタイミングで何かするときにObserver<T>

それ以外はAction1<T>を使うのかなーと思っています。

RxJavaを最低限扱うにはこれぐらい知っておくといいのかなーと思います。

オペレータはここであげたflatMapやMap以外にも値をフィルタするfilterなどがありますので調べてみてください。

たとえばこことか。

Retrofit+RxJavaでものすごく悩んだこと

自分的な本題入ります。

Retrofit+RxJavaでTiqavというサービスのクライアントを作ろうとしたとき、流れるデータが何かさっぱりわからなかったため、めちゃくちゃ悩みました。

なお、このAPIを叩くと、以下のようなレスポンスが帰ってきます。

[
 {"id":"3om","ext":"jpg","height":1442,"width":1036,"source_url":"http://example.com/image1.jpg"}, 
 {"id":"1eb","ext":"jpg","height":171,"width":250,"source_url":"http://example.com/image2.jpg"}
]

定義しているinterfaceがこれで、

public interface TiqavApiService {
    @GET("search/{param}")
    Observable<List<TiqavResponseObject>> GetNewest(@Path("param") String paramater);
}

TiqavResponseObjectはTiqavのレスポンスの1要素

{"id":"1eb","ext":"jpg","height":171,"width":250,"source_url":"http://example.com/image2.jpg"}

を受けとるためのPOJOと考えてください。

そしてObservableを返すクラスがこれ


public class CallTiqavApi {

    public static Observable<List<TiqavResponseObject>> getTiqavNewest(){
        Retrofit rf = new Retrofit.Builder()
                .baseUrl("http://api.tiqav.com")
                .addConverterFactory(JacksonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .build();
        TiqavApiService tas = rf.create(TiqavApiService.class);
        Observable<List<TiqavResponseObject>> observable = tas.GetNewest("newest.json");
        return observable;
    }
}

当初は、Observable<List<TiqavResponseObject>>はListをイテレートしてTiqavResponseObjectを流してくれると思って

Observable<List<TiqavResponseObject>> ob = CallTiqavApi.getTiqavNewest();
ob.map(new Func1<TiqavResponseObject, String>() {
    @Override
    public String call(TiqavResponseObject s){
　　　　//値の加工
    }
})

のようなコードを書いていて赤波線が出てしまうので、ただひたすら悩んでいました。

ぶっちゃけ、Observable<List<TiqavResponseObject>>がどんな意味を持っているのが理解できてないことが一番の問題で、Observableの<T>の部分がそのままオペレータにどんどん渡される、ということを理解したらすぐわかりました。

つまり、TiqavResponseObjectを流して加工するには、

Observable<List<TiqavResponseObject>> ob = CallTiqavApi.getTiqavNewest();
ob.flatMap(new Func1<List<TiqavResponseObject>, TiqavResponseObject>() {
    @Override
    public TiqavResponseObject call(<List<TiqavResponseObject> t){
　　　　return Observable.from(t);
    }
}).map(new Func1<TiqavResponseObject, String>() {
    @Override
    public String call(TiqavResponseObject s){
　　　　//値を加工
    }
})

とすればよかったわけです。

flatMapはObservableを返すクラスに含めてもいいですね。

たぶん間違えてるところとかあると思うのでマサカリお願いします。