【Unity】Subscribeしなくても動作してかつSubscribeすることで結果も必ず受け取れるストリームを作る方法【UniRx】

# はじめに
Rxを学んでいざ自分でストリームの定義をし始めると、Coldなストリームの「`Subscribe()`しないと起動しない」という挙動が使いづらいと感じました。

どうにかならないかといろいろ調べていたらCold→Hot変換用のメソッドと`Subject<T>`の関連クラスにその答えが見つかったので、そのメモです。

「自分でストリーム作ったら公開するときにHot変換するといい」的なことは見たことありましたがあまりピンと来てなかったので、Rx勉強中の誰かの参考になれば幸いです。

# 具体的にどういうことか
例えばアニメーションで何かを表示する`Show()`というメソッドを作ろうとしたときに、

* `Show()`メソッドを呼び出すが完了通知自体は必要ない、つまり呼び出すだけにしたい
* `Show()`メソッドを呼び出して完了通知を受け取りたい、つまり`Subscribe()`するもしくは`WhenAll()`や`Zip()`に渡して待ち合わせしたい

という２つの使い方を考えます。

```Window.cs
using System.Collections;
using UnityEngine;
using UniRx;

public class Window : MonoBehaviour
{
    private RectTransform _rectTransform;

    private RectTransform rectTransform {
        get { return _rectTransform ?? (_rectTransform = this.GetComponent<RectTransform>()); }
    }

    // 呼び出すだけで起動したい、かつSubscribeしたら結果を受け取れるようにしたい
    public IObservable<Unit> Show()
    {
        // NOTE: Coldなストリームを返す、つまり起動にはSubscribeが必要
        return Observable.FromCoroutine(ShowCoroutine);
    }

    private IEnumerator ShowCoroutine()
    {
        this.rectTransform.localScale = new Vector3(0, 1, 1);

        yield return null;

        while (this.rectTransform.localScale.x < 1) {
            this.rectTransform.localScale += new Vector3(0.05f, 0, 0);
            yield return null;
        }
    }
}
```

両方の用途で使えるようにするには戻り値として`IObservable<T>`を返せばいいのですが、普通に書くと前者はストリームの起動のためだけに`Subscribe()`を呼び出す必要があって冗長になってしまいます。

`Observable.FromCoroutine`はColdなストリームを返すからです。

# ダメなやり方

## 内部で`Subscribe()`を呼び出して起動しておく
単純に「内部で`Subscribe()`しちゃえば外側で`Subscribe()`しなくても起動されるじゃん」と思ったのでやってみました。

```Window.cs
    public IObservable<Unit> Show()
    {
        var stream = Observable.FromCoroutine(ShowCoroutine);
        // Subscribeを呼び出して公開前に起動
        stream.Subscribe();
        return stream;
    }
```

これだと`Subscribe()`しない場合の動作はOKですが、`Subscribe()`すると２回起動してしまいます（Coldなストリームの特性）。

ちゃんとHot変換するには`Multicast()`メソッドを使用します。

## `Multicast()`でHot変換後、`Connect()`で起動しておく

```Window.cs
    public IObservable<Unit> Show()
    {
        var stream = Observable.FromCoroutine(ShowCoroutine).Multicast(new Subject<Unit>());
        // NOTE: Publishを使うと短く書ける
        // var stream = Observable.FromCoroutine(ShowCoroutine).Publish();
        stream.Connect();
        return stream;
    }
```

`Multicast(ISubject<T> subject)`は内部で一人代表して`Subscribe(subject)`し、`OnNext(T)`の値を分配することでColdなストリームをHot変換してくれます。戻り値は`IConnectableObservable<T>`で、`Connect()`を呼び出すと内部的な`Subscribe()`を開始、つまりストリームが起動します。

これで`Subscribe()`を呼び出しても２回起動しなくなりましたが、今度は`Subscribe()`される前にストリーム内の処理が終わったときに`OnNext()`が受け取れません。

```WaitForObservable.cs
using System;
using UnityEngine;
using UniRx;

public class WaitForObservable : MonoBehaviour
{
    [SerializeField]
    private Window _window;

    void Start()
    {
        var showStream = _window.Show();

        // WARNING: Subscribe前にShowStreamが完了しているのでDebug.Logは呼ばれない！
        Observable.Timer(TimeSpan.FromSeconds(1.5f)).Subscribe(_ => {
            showStream.Subscribe(__ => Debug.Log("OnNext()"));
        });
    }
}
```

上記の例だと別に`OnNext()`でなくても`OnComplete()`が受け取れるので大丈夫といえば大丈夫なのですが、どうしても取りこぼした`OnNext()`の値がほしい場合は困ります。

# どうすればいいか
`Subject<T>`の亜種で、`OnNext(T)`で発行した値をキャッシュしてくれる`ReplaySubject<T>`を使えばOK。`ReplaySubject<T>`は`Subscribe()`時にこれまで発行したすべての`OnNext(T)`の値を発行します。

```Window.cs
    public IObservable<Unit> Show()
    {
        var stream = Observable.FromCoroutine(ShowCoroutine).Multicast(new ReplaySubject<Unit>());
        // NOTE: Replayを使うと短く書ける
        // var stream = Observable.FromCoroutine(ShowCoroutine).Replay();
        stream.Connect();
        return stream;
    }
```

これでストリームが`Subscribe()`前に終了していても`Subscribe()`時に`OnNext(T)`が発行されます。

`Subject<T>`の亜種は他にもありますが、詳細は参考資料のリンク先を参照してください。

上記の例だと`FromCoroutine`が`OnComplete()`前に一度だけ`OnNext(T)`を発行するので、`AsyncSubject<T>`でもいいですね。

# 注意！
今回は「`Subscribe()`しなくても起動したい」ということで`Dispose()`しなくても問題ないストリームを前提としましたが、基本的にはメソッド呼び出し側が責任を持ってストリームの寿命を管理できるようにすべきです。

例えば`FromCoroutine`に渡すコルーチンが内部的に`while(true){}`でずっとループし続ける場合などは止める手段がないとメモリリークの原因になります。

```Window.cs
using System.Collections;
using UnityEngine;
using UniRx;

public class Window : MonoBehaviour
{
    public IObservable<Unit> Show()
    {
        var stream = Observable.FromCoroutine(ShowCoroutine).Replay();
        // WARNING: disposable.Dispose()を呼ばないとコルーチンが停止しない
        var disposable = stream.Connect();
        return stream;
    }

    private IEnumerator ShowCoroutine()
    {
        while (true) {
            yield return null;
        }
    }
}
```

より安全にHot変換を行うには`RefCount()`をセットで使いましょう。

`RefCount()`を使うと`Subscribe()`が呼ばれたときにストリームが起動し、`Subscribe()`したObserverが全員`Dispose()`した時点でストリームがキャンセルされます。

`Observable.FromCoroutine`の場合はObserverが全員`Dispose()`されたらコルーチンが停止します。

```Window.cs
    public IObservable<Unit> Show()
    {
        return Observable.FromCoroutine(ShowCoroutine).Replay().RefCount();
    }
```

ただ、上記の場合はストリームを起動するのに少なくとも１回は`Subscribe()`が呼ばれる必要があり、この記事の冒頭に戻ってしまいますね。

結局は「`Subscribe()`を必ず呼ばせて寿命管理を呼び出し側の責任とする」のが正しいのかもしれません。

## <font color="red">2017/7/6 追記</font>
「`Subscribe()`しなくても実行されるストリーム」というのはほかのオペレータと組み合わせたときに待って実行することができないのでそもそも期待される実装とは異なることがほとんどです。この記事の内容は参考程度でお考えください。

# 参考資料
* [Rx入門 (6) - Subject<T>の亜種](http://blog.xin9le.net/entry/2011/12/22/021640)
* [UniRx オペレータ逆引き - ストリームの分岐](http://qiita.com/toRisouP/items/3cf1c9be3c37e7609a2f#%E3%82%B9%E3%83%88%E3%83%AA%E3%83%BC%E3%83%A0%E3%81%AE%E5%88%86%E5%B2%90)
* [UniRx オペレータ逆引き - Subject系](http://qiita.com/toRisouP/items/3cf1c9be3c37e7609a2f#subject%E7%B3%BB)

# 感想
「Cold→Hot変換よくわからんなぁ」「Subjectの亜種も用途がよくわからんなぁ」と思って放置してたがやりたいことを調べていく上で理解できたので良かった。

次はRxの非同期処理周りを理解したい。