Rxで知っておくと便利なSubjectたち

この記事は[RxJava Advent Calendar 2015](http://qiita.com/advent-calendar/2015/rxjava) 11日目の記事です。

RxJava・RxAndroidの紹介記事の中にはよく「面倒くさいASyncTaskを使わなくて済む」といったRxJavaを非同期処理の代わりとして紹介されています。
もちろん、非同期処理を置き換えられるという点はRxJavaの有効的な使い方の一つですが、RxJavaには他にも便利に使える機能があります。
今回はそれらの機能の一つであるSubjectについて紹介したいと思います。

# Subjectとは

短く言うと「`Subscriber`と`Observable`の2つの機能を併せ持ったもの」です。
`Subscriber`にあるような`onNext`や`onError`、`onComplete`といったメソッドを呼び出せ、`Observable`のように`subscribe`メソッドを呼び出すことができます。

```Subject1.java
Subject<String, String> subject = BehavorSubject.create();
subject.onNext("Hoge");
subject.subscribe(System.out::println); // "Hoge"が標準出力へ出力される
```

`Observable.create()`を使ったObservableの生成方法だと、任意のタイミングで`onNext`などを呼び出すことが非常に面倒になってしまいますが、SubjectではSubject自身に対して`onNext`などを呼ぶことができるため、任意のタイミングでの呼び出しが簡単にできます。

# Subjectの種類と特徴

Subjectには、`onNext`で渡された値をどのようにするかで複数種類の物があります。
それぞれを使い分けることで、今まで複雑になりがちだった記述を短くきれいに書くことができるようになります。

## AsyncSubject

<img width="640" alt="S.AsyncSubject.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/92d3d793-f883-98ff-8681-0f17438abfcf.png">

AsyncSubjectでは、AsyncSubject側の`onCompleted`が呼ばれた直後に、最後に`onNext`で渡された値のみがSubscriberの`onNext`の引数として渡されます。

<img width="640" alt="S.AsyncSubject.e.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/631fa0fd-d918-2685-f1c2-58071068613f.png">

途中でAsyncSubject側の`onError`が呼ばれた場合は、Subscriberの`onNext`は呼ばれずに`onError`が呼ばれます。

### 使いどころ

onNextが呼ばれるのが最後の要素だけで、onCompletedを呼ばないと要素が流れてこないというのが少し独特ですが、値が一つしか流れてこないor最後の一つしか必要でない時に、コールバックを受け取る実装をするのに便利です。
例として以下のシチュエーションをAsyncSubjectで実装してみたいと思います。（実装は基本的にRxAndroidを使用しています。）

> 画面が表示されるより前にデータ取得を開始する。
画面が表示された時にデータ取得が終わっていればそのデータを表示する。
終わっていなければロード中のインジケーターを表示して、データ取得が完了し次第取得したデータを表示する。

```Initializer.java
// Applicationクラス(HogeApplicationとする)のonCreateでインスタンス化しておく。
public class Initializer {
    public AsyncSubject<Data> asyncSubject = AsyncSubject.create();

    public Initializer(){
        generateObservable()
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(data -> {
                    asyncSubject.onNext(data);
                    asyncSubject.onCompleted();
                });
    }

    public Observable<Data> generateObservable(){
        return Observable.create(subscriber -> subscriber.onNext(longTask()));
    }
    
    public Data longTask() {
        //時間のかかる処理 (ネットワークからのデータ取得など)
    }
}
```

```SubActivity.java
// MainActivityでボタンをタップしたりした後に呼ばれると仮定
public class SubActivity extends RxActivity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_sub);

        // ここでダイアログを表示させる
        AlertDialog dialog = ......;
        dialog.show();

        // HogeApplication#getInitializer() が上のInitializerのGetter
        ((HogeApplication) getApplication()).getInitializer().asyncSubject
                .compose(bindToLifecycle()) // AndroidのライフサイクルにBind
                .subscribe(data -> {
                    // dataをリストに表示したりする
                    dialog.dismiss();
                }, throwable -> {
                    // エラーが出たことを表示する
                    dialog.dismiss();
                });
    }
}
```

画面を表示するより前にデータが来てしまった時と、データが来なかった時両方に一つのコードで対応することができるようになりました。

## BehaviorSubject

<img width="640" alt="S.BehaviorSubject.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/9322a7ef-325a-068f-7202-ddf39aa54ccb.png">

直前に`onNext`で渡された値を保持し、`subscribe()`した直後に保持していた値を流します。その後の動作は後述の`PublishSubject`と同等です。

<img width="640" alt="S.BehaviorSubject.e.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/f57cb952-f802-243c-f98e-1dafb6bcb359.png">

一度`onError`で例外が流されると、`onNext`は無視され、`subscribe()`後にSubscriber側の`onError`が呼ばれます。

### 使いどころ

直前に`onNext`で渡したデータがやってくるという点で、使いどころは少し限定されていますが、[RxLifecycle](https://github.com/trello/RxLifecycle)で、ActivityやFragmentのライフサイクルを取得・判別する部分で使われているのが一番しっくり来た使い方でした。

[RxLifecycleの紹介](http://qiita.com/hide92795/items/02533c8593ce51070b57)

```RxLifecycle.java
return new Observable.Transformer<T, T>() {
    @Override
    public Observable<T> call(Observable<T> source) {
        return source.takeUntil(
            Observable.combineLatest(
                lifecycle.take(1).map(correspondingEvents),
                lifecycle.skip(1),
                new Func2<R, R, Boolean>() {
                    @Override
                    public Boolean call(R bindUntilEvent, R lifecycleEvent) {
                        return lifecycleEvent == bindUntilEvent;
                    }
                })
                .onErrorReturn(RESUME_FUNCTION)
                .takeFirst(SHOULD_COMPLETE)
        );
    }
};
```

上のコードは、RxLifecycleの[RxLifecycle.java](https://github.com/trello/RxLifecycle/blob/master/rxlifecycle/src/main/java/com/trello/rxlifecycle/RxLifecycle.java)より一部抜粋・修正を加えたものです

`source`はライフサイクルを適用させるObservable。
`correspondingEvents`は、ActivityやFragmentのイベントに対して、対のイベントに変換を行うFunction（例：`OnCreate`ならば`OnDestroy`に変換）。
`lifecycle`は、対象のAcitvityのライフサイクルの状態が流れてくる**BehaviorSubject**です。

コードを見ると、`lifecycle.take(1).map(correspondingEvents)`でバインドしたライフサイクルに対する終わりのイベントを取得し、`lifecycle.skip(1)`と`combineLatest`でつなげることで、`lifecycle`に新しい値が流れてくるごとに評価が行われます。
この評価で、終わりのイベントと次のイベントが同じ（終わりのライフサイクルに到達した）時に`takeUntil`で元のObservableを途中で止めています。

この、ライフサイクルイベントのストリームとしてBehaviorSubjectが使われています。前の状態をキャッシュしてくれているので、subscribe時に直前に起きていたイベントを取得できるようになるわけです。
(onCreateからonStartの間に`bindToLifecycle`した場合は、`lifecycle`にonCreateのイベントがキャッシュされている)

## PublishSubject

<img width="640" alt="S.PublishSubject.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/4d085045-e2e8-b7e3-50de-d31c700e60bf.png">
<img width="640" alt="S.PublishSubject.e.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/ff2c49ad-6314-faa7-f095-1f494b96b89f.png">

Subjectの中で、一番動作の理解がしやすいSubjectです。Subjectへの`onNext`,`onError`, `onSuccessed`すべての呼び出しが、Subscriber側の同じメソッドにそのまま伝搬します。言ってしまえば、普通のリスナーと同じような挙動をします。

### 使いどころ

上で言ってしまったように、リスナー系の仕組みをRxを使用して置き換えるようなときに使用することができます。

例として、ArrayAdapterなどでリストを表示するときに、クリックイベントを元のActivityやFragmentへ返す動作を実装してみます。

```HogeAdapter.java
public PublishSubject<Integer> clickObservable = PublishSubject.create();

@Override
public View getView(int position, View view, ViewGroup parent) {
    ViewHolder holder;
    if (view == null) {
        view = inflater.inflate(R.layout.view_list_item_hoge, null);
        holder = new ViewHolder(view);
        view.setTag(holder);
    } else {
        holder = (ViewHolder) view.getTag();
    }

    Item item = getItem(position);
    holder.text.setText(item.getData());
    view.setOnClickListener(v -> clickObservable.onNext(history));
    return view;
}
```

```HogeActivity.java
public HogeActivity extends RxActivity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_hoge);

        // ここでダイアログを表示させる
        ListView list = (ListView) findViewById(R.id.list_hoge);
        HogeAdapter adapter = new HogeAdapter();
        adapter.clickObservable
            .compose(bindToLifecycle())
            .subscribe(position -> {
            // position番目の要素がクリックされた時の動作
        });
    }
}
```

これで、HogeActivity側でクリックされたポジションのイベントを取得できるようなりました。

## ReplaySubject

<img width="640" alt="S.ReplaySubject.png" src="https://qiita-image-store.s3.amazonaws.com/0/29459/7d159981-6940-cf30-9634-30bcc130d6f5.png">

大きな特徴は、`subscribe()`した後に、Subjectに今まで`onNext`で流されていた値が全てやってくるという点です。
今までのデータがすべて流れてきたあとの動作はPublishSubjectと同等です。

### 使いどころ

元データからの差分をObservableに流すような機構と相性が合う気がします。
また、`skip`すれば任意の場所の値を取れるという部分を活用すれば、Listの代わりとして使えそうでもあります。

## SerializedSubject

今までのSubjectはすべて、Subject側の`on***`メソッドを同じスレッドから呼び出す必要がありました。
もし、複数のスレッドから`on***`メソッドが呼ばれる場合は、このSerializedSubjectでラッピングしてあげる必要があります。

```java
PublishSubject<Data> subject = PublishSubject.create();
SerializedSubject<Data, Data> serializedSubject = new SerializedSubject(subject);
```

# おわりに

Subjectはとてもクセのある種類がありますが、使い分ければそれぞれで便利に使うことができます。
なにかミスなどありましたらコメントにて教えていただければと思います。

# ライセンス表記
この記事で使っている画像はすべてReactiveXの著作物であり、[ReactiveX - Subject](http://reactivex.io/documentation/subject.html) のページから引用したものです。
画像のライセンスは[Creative Commons Attribution 3.0](https://creativecommons.org/licenses/by/3.0/)の下、表示しています。
RxLifecycleより抜粋したコードは[ApacheLicense Version 2.0](http://www.apache.org/licenses/LICENSE-2.0)の下で公開されます。
それ以外の私が制作したコードはPublicDomainの下で公開します。