【UniRx】IObservableのHotとColdを完全に理解する

#はじめに

RxにつきもののHot Coldについて解説します。

HotColdについてはすでに様々な記事で解説がなされているとは思いますが、直感的には理解ができても、本質的な部分については理解できないことがありました。
先日、ようやく納得がいくまで本質部分を理解できましたので、記事にしてまとめます。
一応、他の記事とは違った視点での説明になっていると思います。

ちなみに、Hot変換については別記事で書いていますのでこちらを参照ください。ただし、本記事の知識が前提となっています。

https://qiita.com/yutorisan/items/68bd68bbffe3c4f3ba32

##その前に`Subject<T>`クラスについて
Hot-Coldと`Subject<T>`クラスには深い関係があります。
そのため、Hot-Coldの説明の前に`Subject<T>`クラスについてもう一度復習します。

御存知の通り、`Subject<T>`クラスは`IObserver<T>`インターフェイスと`IObservable<T>`インターフェイスを両方実装したクラスです。
したがって、`IObserver`としても`IObservable`としても振る舞うことができるという特徴があります。

###複数の購読者を管理する機能を持つ

ここで重要なのは、`Subject`クラスは、__`Subscribe`してきた購読者（つまり`IObserver`）を内部にリストとして保持する__ということです。

購読者が`Subject`クラスの`Subscribe`メソッドを呼び出すと、`Subject`クラスが内部に持つ`IObserver`のコレクション（購読者リスト）に、引数で渡された`IObserver`を追加します。

![Subjectの役割_Subscribe.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/482d0155-0f0e-b1de-7ae2-8da1930c6b5c.png)


そして、`Subject`クラスの`OnNext`や`OnError`、`OnCompleted`が呼ばれたときに、保持している購読者リストにあるすべての購読者に対して、同様の通知を発行します。

![Subjectの役割_OnNext.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/3aa1469b-fe62-90d4-97b7-38a846bc2df7.png)


つまり、`Subject`クラスとは、__1つのObservableシーケンスを複数の購読先に分配する、マルチキャストするという役割を持っているクラス__なのです。

まとめると、`Subject`クラスの特徴は次のとおりとなります。

- __複数の購読者を管理する機能を有する__
- __1つのObservableシーケンスをすべての購読先にマルチキャストする__

このことを踏まえた上で、Hot Coldの説明をします。

#HotとCold

##Hot Observableとは

__Hotな`IObservable`__とは、__購読者を複数持つことができる`IObservable`__のことを指します。
そして、Hotな`IObservable`は「__1つのObservableシーケンスを、すべての購読者に対してマルチキャストする__」という特徴があります。

「複数の購読者を持つ」「1つのObservableシーケンスを複数の購読先にマルチキャストする」と言えばもう__`Subject`クラス__しかありませんね。

###Hot Observableは「Subject」

実は、__Hot Observableとは、`Subject`クラスそのものを指します__。文字で説明されても意味がわからないと思うので、次のコードを見てください。

```csharp
public class HotObservableProvider
{
    private Subject<int> subject = new Subject<int>();

    public void Fire()
    {
        subject.OnNext(1);
        subject.OnNext(2);
        subject.OnNext(3);
    }

    //「HotObservable」から取得できるIObservableはSubjectクラスそのものを指す
    public IObservable<int> HotObservable => subject;
}
public class User
{
    public User()
    {
        HotObservableProvider provider = new HotObservableProvider();
        provider.HotObservable //このIObservable<int>はSubjectクラスそのものを指すのでHot
                .Subscribe(n => { /*何らかの処理*/ });
    }
}
```

この`HotObservableProvider`クラスが提供する`HotObservable`プロパティから取得できる`IObservable<int>`インターフェイスは、__その実体がまさに`Subject<int>`クラスそのものを指しています__。
このように、実体が`Subject`クラスそのものを指す`IObservable`は、Hotであると考えることができます。

また、実体が`Subject`そのものではなくとも、__`Subscribe`したときに実質的に`Subject`を`Subscribe`したのと同じような効果があるような`IObservable`__も、Hotと見做すことができます。これについても具体例を用意しました。

```csharp
//IObservableの実体はSubjectクラスではない
public class HotObservable : IObservable<int>
{
    private Subject<int> subject = new Subject<int>();

    public void Fire()
    {
        subject.OnNext(1);
        subject.OnNext(2);
        subject.OnNext(3);
    }

    //Subscribeしたとき、SubjectクラスをSubscribeするのと同じ効果がなる
    public IDisposable Subscribe(IObserver<int> observer)
    {
        return subject.Subscribe(observer);
    }
}
public class User
{
    public User()
    {
        IObservable<int> observable = new HotObservable();
        //「observable」の実体はSubjectではないが、
        // Subscribeすると実質SubjectをSubscribeしてるのと同じ
        observable.Subscribe(n => { /*何らかの処理*/ });
    }
}
```

先ほどの例と違って、`HotObservable`クラス自体が`IObservable`となっています。`User`が`HotObservable`クラスを`IObservable`として使用するとき、当然この`IObservable`の実体は`HotObservable`であり`Subject`ではありません。しかしながら、`HotObservable`クラスは、__`Subscribe`すると内部に持つ`Subject`を`Subscribe`するのと同じ効果となるように実装されています__。したがって、この場合もHot Observableであると見做すことができます。

まとめると、Hot Observableは、__購読先が`Subject`であるような`IObservable`__とも言えると思います。


##Cold Observableとは
対して__Coldな`IObservable`__とは、__購読者を1つしか持つことができない`IObservable`__のことを指します。

Coldな`IObservable`は購読者を1つしか持てないので、1つの購読者に対して常に専属の`IObservable`が存在するということになります。Hotな`IObservable`は__`Observable:Observer = 1:多`__であったのに対して、Coldな`IObservable`は__`Observable:Observer = 1:1`__となることが特徴です。

なぜCold Observableが「購読者を1つしか持てない」のかというと、__`Subject`クラスを利用していないから__です。`Subject`クラスを利用していないので、複数の購読者を管理することができず、ただ1つの購読者に対して専属のObservableが必要になってしまいます。
1つのObservableシーケンスを共有しない（つまり購読者が1つだけ）ならばColdのままで問題ないのですが、共有したい（つまり購読者が複数存在する）場合はHot変換が必要になる場合がある、というのはこのためです。

ちなみに、Rxのオペレータやファクトリメソッドから出てくる`IObservable`は、そのほとんどが`Subject`を使っていません。そのため、普通にRxを使っている分には、ほとんどがColdな`IObservable`になります。

##まとめると…

- __Hot Observable__
    - 購読先が`Subject`であるような`IObservable`
    - 複数の購読者を管理可能
    - `Observable:Observer = 1:多`
- __Cold Observable__
    - 購読先が`Subject`ではないような`IObservable`
    - ただ1つの購読者しか管理できない
    - `Observable:Observer = 1:1`




#Observableソースのお話

これで終わりではなく、大事な__Observableソース__についてのお話があります。

Observableソースとは、__Observableシーケンスに流れてくる値の「発生源」__です。
そしてこのObservableソースにも__Hot__と__Cold__が存在し、この違いを認識していないと思わぬバグの原因となりますので、Rxを使う上では必ず理解する必要がある知識です。

##Observableソースの例

ObservableソースはObservableシーケンスに流れてくる値の発生源などと言われてもいまいちピンとこないと思いますので、例を紹介します。

###`Subject<T>`がObservableソース

`Subject<T>`クラスがObservableソースとなる例です。

```csharp

Subject<int> subject = new Subject<int>();

subject.Subscribe(n => { /*何らかの処理*/ });

//ここでSubjectにOnNextすることによってObservableシーケンスに値が乗る
//→SubjectがObservableソース
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
```

`Subject`に対して直接値を`OnNext`によって送出しています。この場合、Observableシーケンスを流れる値の発生源はこの`OnNext`にあるので、Observableソースは`Subject`であることが言えます。

###ファクトリメソッドがObservableソース

続いて、ファクトリメソッドがObservableソースとなる例です。

```csharp
//ファクトリメソッドから値が送出されている
//→ファクトリメソッドがObservableソース
Observable.Range(1, 10)
          .Subscribe(n => { /*何らかの処理*/ });
```

この場合、`Observable.Range`からObservableシーケンスの値が送出されていることは言うまでもなく、Observableソースはこのファクトリメソッドということが分かります。



##ObservableソースのHotとCold

ObservableソースにもHotとColdがあると書きました。
察しが良い方は勘付くかもしれませんが、上の例で紹介した__`Subject<T>`がObservableソースのものがHot__で、__ファクトリメソッドがObservableソースのものはCold__です。理由は、`Subject<T>`は複数の購読者を管理できるのでHot、ファクトリメソッドは`Subject<T>`を使っていない[^2]ので複数の購読者を管理できずCold、ただそれだけです。

[^2]: 一部例外もあるようです。詳しくは@acple@githubさんの[こちらの記事](https://qiita.com/acple@github/items/8d3a4d3414fa59adff70)を参照ください。

では、ObservableソースのHotとColdが違うとどう影響してくるのかを説明します。

###HotなObservableソース

HotなObservableソースとは、__`Subject<T>`が発生源となるObservableソース__のことを指します。

HotなObservableソースには次のような特徴があります。

- 何度`Subscribe`しても同一のObservableシーケンス
- シーケンスの途中から`Subscribe`した人には途中からしか流れてこない

![Hot Observableソース.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/4717e5a6-5e2b-d4bf-bb22-fda609e30af5.png)

これはまぁ、想像通りというか、詳しい説明はいらないと思います。
ところが、次のColdなObservableソースの場合、注意が必要です。

###ColdなObservableソース

ColdなObservableソースとは、HotではないすべてのObservableソースを指します。
Rxに用意されているファクトリメソッドから作られる`IObservable`は、そのほとんどが内部で`Subject<T>`を使用していないため、ColdなObservableソースとして動作します。

ColdなObservableソースには次のような特徴があります。

- __`Subscribe`するたびに新しいObservableシーケンス__
- __購読者一人ひとりに対して専用のObservableシーケンス__なので、__途中からという概念が存在しない__

HotなObservableソースと決定的に違う点は、下図にように「__`Subscribe`するたびに新しいObservableシーケンスが生成される__」という点です。

![Cold Observableソース.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/f0c06e48-a031-37d5-aa54-96f2ddbcf37e.png)

Hotの場合と違って、ColdなObservableソースは`Subscribe`するたびに、新しくObservableシーケンスが生成されています。一体なぜこのような動作になるのでしょうか？

答えは、「__内部に`Subject`を使っていないから__」となります。`Subject`を使わないということは、購読者に対して「__直接__」値を発行することを意味します。ちょっと何を言っているのかわからないと思うので、コードで説明します。ここでは、`Observable.Range`ファクトリメソッドの動作を簡易的に実装した独自の`MyObservable.MyRange`ファクトリメソッドを作成してみました。

```csharp
public static class MyObservable
{
    public static IObservable<int> MyRange(int start, int end) =>
        new MyRange(start, end);
}
class MyRange : IObservable<int>
{
    private int start, end;
    public MyRange(int start, int end)
    {
        this.start = start;
        this.end = end;
    }

    public IDisposable Subscribe(IObserver<int> observer)
    {
        //Subscribeしてくれた人に直接値を送出する
        for (int i = start; i <= end; i++)
        {
            observer.OnNext(i);
        }
        observer.OnCompleted();
        return new Subscription();
    }

    class Subscription : IDisposable
    {
        public void Dispose()
        {
            //SubscribeしたらすぐOnCompletedになるので何もしない
        }
    }
}
```

`Subject`を使わずに値を発行するには、`Subscribe`の引数に渡された`IObserver`（購読者）に対して__直接`OnNext`するしかない__のです。必然的に、このObservableシーケンスは、__購読してきた`IObserver`専用__となってしまうのです。したがって、複数回`Subscribe`されれば、その分新しくObservableシーケンスを生成せざるを得なくなります。これが、上記の図の説明となります。

そして当然、このような仕様を理解していないと、想定とはまったく異なる動作をしてしまう危険性があります。例えば、以下のように1つの`Observable.Interval`を時間差で複数回`Subscribe`した場合。

```csharp
//0.1秒ごとに0,1,2,…とインクリメントする値を発行するColdなObservableソース
IObservable<long> observableTimer = Observable.Interval(TimeSpan.FromSeconds(0.1));

observableTimer.Subscribe(n => Debug.Log(n));
await Task.Delay(1000); //1秒差でColdなObservableソースを2回購読
observableTimer.Subscribe(n => Debug.Log(n));
```

「同じ変数`observableTimer`を2回`Subscribe`してるんだから、Observableシーケンスは同じで、2個めの`Subscribe`には10,11,12,…って値が来るんでしょ」と思っていたら、__大間違い__です。
`Observable.Interval`はColdなObservableソースなので、`Subscribe`するたびに新しく専用のObservableシーケンスが生成されるので、両方とも0,1,2,…と値が発行されます。
もし、前者のような動作をしてほしい場合はHot変換という作業が必要になります。

このように、ObservableソースのHot-Coldを意識することは非常に重要で、これらの違いを理解しないままRxを使い込むと、バグの原因となりますので十分に注意が必要です。

##まとめると…

- `Subject`から発生するObservableソースはHot
- ファクトリメソッドを始めとした、`Subject`以外から発生するObservableソースはCold
- ColdなObservableソースは、`Subscribe`するたびに新しく専用のObservableシーケンスが生成される
- 想定する動作に合わせて適切にHot変換が必要


#さいごに

この記事では、`IObservable`のHot-ColdとObservableソースのHot-Coldを分けて記述しましたが、本質的には（おそらく）同じで、`Subject`を使っているか、使っていないか、それだけの違いになります。
ただその小さな違いが、大きな違い（バグ）を生み出す原因になるので、十分に注意してRxを使うことが必要です。

ご意見ご指摘等ありましたら是非コメント宜しくお願いします！

[2021/4/2追記]
Hot変換の記事を書きましたので、ぜひこちらも参照ください！

https://qiita.com/yutorisan/items/68bd68bbffe3c4f3ba32

#参考記事

https://qiita.com/toRisouP/items/f6088963037bfda658d3

https://qiita.com/acple@github/items/8d3a4d3414fa59adff70


[^1]: 正確に言うと、「実体が`Subject<T>`クラスである`IObservable<T>`インターフェイス」もしくは「`Subscribe`メソッドを呼び出した先で`Subject<T>`クラスの`Subscribe`を呼び出しているような`IObservable<T>`インターフェイス」となります。`IObservable<T>`インターフェイスの「実体」が`Subject<T>`クラスであるものはHot Observableである、というと少し語弊があります（Hot変換の`Publish`オペレータで出てくる`IObservable<T>`の実体は`ConnectableObservable`クラスですし）が、直感的にわかりやすい表現にするためにこのような表現としました。