【UniRx】Cold→Hot変換の代表、「Publishオペレータ」を完全に理解する

#はじめに

ReactiveExtensions（以下Rx）の鬼門はHot-Coldの概念とHot変換だと思います。

- Hot変換が必要なのはストリームを枝分かれさせたいときである…
- Hot変換するには`Publish()`して`Connect()`すれば良い…

等々はいろいろな記事で紹介されていますが、では`Publish()`すると裏で一体何が起こるのか？`Connect()`すると何が起こる？というのは、あまり詳しく解説されている記事が少ない印象です。

そこで、本記事ではRxの鬼門`Publish()`メソッドを「とりあえず」で使うのを卒業し、何が行われているかを「完全に理解」できるように解説していきます。

#[前提知識]HotとCold

まずは前提となるHotとColdの知識が必要です。
別記事で書きましたので、もしHotColdの知識が曖昧な方はご覧ください。
`Subject<T>`クラスの役割とHotColdとの関係性についても解説しています。

https://qiita.com/yutorisan/items/844eaeab392abf03ce80

↑の記事の内容を要約すると次の３つに言及しています。

###1.`Subject<T>`クラスの特徴について
`Subject<T>`クラスは__1つのObservableシーケンスを複数の購読者に対してマルチキャストする機能__を持つ。

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

###2.Hotな`IObservable`とColdな`IObservable`の違いについて

Hotな`IObservable`とは、__`Subject<T>`クラスから出ている`IObservable`__のことを指す。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/ac3152cf-fdff-1a5d-a4af-3dab74c12b6f.png)


実体は`Subject<T>`クラスなので、複数の購読者に対してObservableシーケンスをマルチキャストできるという特徴を持つ。

対して、__それ以外の（`Subject`以外から出ているような）`IObservable`はCold__と呼ばれる。
UniRxのオペレータから出てくる`IObservable`はほぼColdである。

###3.Observableソースについて

Observableソースとは、Observableシーケンスに流れてくる__値の発生源__のこと。これにもHotとColdがある。

HotなObservableソースは、何回`Subscribe`されてもObservableシーケンスは１つ。`Subscribe`した時点以降の通知が届くようになる。
__`Subject`がObservableソースの場合は、HotなObservableソース__となる。

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

対してColdなObservableソースの場合、`Subscribe`されたらその都度専用のObservableシーケンスが生成される。
専用のObservableシーケンスなので、いつ`Subscribe`してもすべての通知が届く。
主に__ファクトリメソッド等から生成される場合、ColdなObservableソース__となる。

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

これらの内容を踏まえた上で、`Publish`オペレータについて解説していきます。

#Cold→Hot変換するにはどうすればよいのか

前述したHot, Coldの定義と、`Subject<T>`クラスの特徴を踏まえて、__Cold__な`IObservable`を__Hot__な`IObservable`に変換するにはどうすれば良いのかを考えます。
要は、__`Subject`クラスから出ている`IObservable`ならば確実にHot__なのですから、「__ColdなObservableを`Subject`クラスから再発信すれば良いのではないか__」と考えることができます。

具体的には、次の手順を踏みます。

###1.`Subject<T>`クラスを用意します。

![Subjectクラス.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/08874ecf-d7fa-f6ab-f994-ec58b51f2ed0.png)

###2. `Subject<T>`は、Hot変換したいColdな`IObservable<T>`を代表して1回だけ`Subscribe`します。

![2.ColdをSubscribe.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/7b999aac-a7aa-e733-b11c-b69dfb16819d.png)


###3. `Subscribe`すると、`Subject<T>`に対してColdな`IObservable<T>`から通知が流れてきます。

![3.通知が流れてくる.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/441cabb5-f96a-bea0-45db-4b2e5c6498b9.png)


###4. `Subject<T>`クラスの`IObservable<T>`を外部に公開します。
__`Subject<T>`クラスから出ている`IObservable<T>`はHotな`IObservable<T>`__なので、Hot変換が完了しました。

![4.Hot変換完了.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/675d61cd-60db-d14b-114a-771388f2c262.png)


`Subject<T>`クラスを使うことで、無事にHot変換を行うことができました。


#`Publish`すると何が起こる

Hot変換する手順は上記の通り。じゃあ`Publish`オペレータは上記のことをやっているんだね！と思ったら、実はそうではありません。なんと、__`Publish`オペレータだけではObservableシーケンスに対してなんの影響も一切与えません__。つまり、`Publish`オペレータをつけただけでは、Hot変換もまだ完了しないのです。

じゃあ`Publish`オペレータは何をしているのか？というと、いつでもHot変換ができるような__「準備」__をしています。

##`Publish`は「準備」するだけ

`Publish`オペレータが実際に行っている処理は、次に示す「準備」のみに留まります。

- __Hot変換用の`Subject<T>`クラスを用意する。__
- __Hot変換したい元となるColdな`IObservable`を用意する。__

用意しているだけで、__実際に`Subject<T>`や`IObservable<T>`に対して何らかの操作をしているわけではない__ことがポイントです。

##準備したものに対する「操作」は次の工程に委託する
実際にHot変換が完了するには、あと残り2つの操作を行うことが必要です。

1. `Subject<T>`に変換元のColdな`IObservable`を`Subscribe`させる
2. `Subject<T>`クラスの`IObservable<T>`を`Subscribe`する（←__Hot Observableに変換完了__）

これら2つの操作が完了したとき、はじめてHot変換が完了したことになります。
しかし、これらの操作は`Publish`内では行われず、次の工程に委託されます。具体的には、__`Publish`オペレータの戻り値として、これら2つの操作のみが可能なインターフェイス返す__ことで、次の工程への委託を実現します。
そのインターフェイスこそが、__`IConnectableObservable<T>`インターフェイス__です。


##`IConnectableObservable<T>`インターフェイス

`IConnectableObservable<T>`インターフェイスとは、その名の通り__`Connect`ができる`IObservable<T>`インターフェイス__です。
定義もまんま、`Connect`ができるようになった`IObservable<T>`です。

```csharp
public interface IConnectableObservable<T> : IObservable<T>
{
    IDisposable Connect();
}
```

つまり、`IConnectableObservable<T>`インターフェイスは`Connect`と`Subscribe`のみができるインターフェイスということになります。

そして、前項で説明したHot変換に必要な2つの操作が、それぞれ`Connect`と`Subscribe`に割り当てられています。
それぞれのメソッドを呼び出すことで、各操作が完了してHot変換が完了することになります。

| メソッド名 | 割り当てられた操作 |
|:-:|:-|
| `Connect`  | `Publish`が用意した内部の`Subject<T>`に、変換元のColdな`IObservable`を`Subscribe`させる  |
| `Subscribe`  | `IObservable<T>`として公開された`Subject<T>`を`Subscribe`する  |

図で表すと次のようになります。

![ConnectableObservable.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/4886bb5b-ea4d-2ad8-0dfa-2afeae1fd834.png)

次項から更に詳しく解説していきます。

#`IConnectableObservable<T>`インターフェイス

`Publish`オペレータの戻り値は`IConnectableObservable<T>`インターフェイスであるということ、そして`IConnectableObservable<T>`インターフェイスは`Connect`と`Subscribe`の2つの操作が可能であることを説明しました。
まだピンとこないと思うので、まずはいくつかのHot変換の実例に則って`IConnectableObservable<T>`インターフェイスの使い方を説明します。

##`IConnectableObservable<T>`の使い方

###例１：いくつかのオペレータを追加した`IObservable`をHot変換して使い回したい

オペレータはColdなので複数回`Subscribe`されると無駄な処理が走ることになります。
それを防ぐためにHot変換を行う例です。

```csharp
//Observableソースを作る
Subject<int> subject = new Subject<int>();

//Subjectからのint型の通知に対して何らかのオペレータを追加
//このまま複数回Subscribeされるとオペレータが重複生成されて無駄な処理が走る
IObservable<TimeSpan> sourceColdObservable =
    subject.Do(n => Debug.Log($"{n}が通知されました")) //Hot変換されていることを確認するためのログ
           .Where(n => n > 0)
           .Select(n => TimeSpan.FromSeconds(n));

//ColdなIObservale「sourceColdObservable」をHot変換する準備をする
IConnectableObservable<TimeSpan> connectableObservable =
    sourceColdObservable.Publish();

//Publish内部のSubjectにsourceColdObservableをSubscribeさせる
connectableObservable.Connect();

//Publish内部のSubject(つまりHotなIObservable)をSubscribeする
connectableObservable.Subscribe(time => Debug.Log($"購読者1:{time}"));
connectableObservable.Subscribe(time => Debug.Log($"購読者2:{time}"));

subject.OnNext(30);
subject.OnNext(60);
subject.OnNext(-20);
subject.OnNext(90);
```

###結果

複数回`Subscribe`しても`Do`オペレータは`OnNext`の分だけ通過してることがわかるので、Hot変換が成功していることがわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/ae0fef4c-eac0-d3f8-53c2-2624b3403c39.png)

####Hot変換しないと…
ちなみにHot変換しないとこんな感じになります。
`Do`オペレータが`Subscribe`の数だけ呼ばれていることからも分かる通り、オペレータ処理が重複実行されてしまっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/4437a5bd-22c9-933c-3acf-647893415ef9.png)


###例２：ファクトリメソッドから生成されたObservableシーケンスを使い回したい

ファクトリメソッドから生成された`IObservable`は基本的にColdなので、`Subscribe`するたびに新しいObservableシーケンスが生成されます。
１つのObservableシーケンスを複数回`Subscribe`して使い回したいためにHot変換を行う例です。

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

//Hot変換の準備をする
IConnectableObservable<long> connectableObservable =
    observableTimer.Publish();

//Publish内でobservableTimerをSubscribe
//→Observableシーケンスが生成され、タイマーが開始される
connectableObservable.Connect();

//Publish内のSubjectをSubscribe
connectableObservable.Subscribe(l => Debug.Log($"最初からの購読者【{l}】"));
//2.5秒後にもう一度Publish内のSubjectをSubscribe
await Task.Delay(2500);
connectableObservable.Subscribe(l => Debug.Log($"2.5秒遅れの購読者【{l}】"));
```

####結果

Hot変換することで、「最初からの購読者」、「2.5秒遅れの購読者」に対して、同一の値が発行されていることが分かります。したがって、1つのObservableシーケンスを共有できていることがわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/37274cb0-c238-a032-60a1-f2996c54a523.png)

####Hot変換しないと…
Hot変換しないと、「2.5秒遅れの購読者」に対しては新しく0から値が発行されていることがわかります。つまり、それぞれObservableシーケンスが別個に生成されているということになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/92cc9b72-8825-49cd-25b4-abcdf3852d53.png)

##なぜ`Connect`が必要なのか

ところで、上記の2例ではどちらも

1. Cold Observableを`Publish`
1. 出てきた`IConnectableObservable<T>`を`Connect`
1. `IConnectableObservable<T>`を好きなだけ`Subscribe`

の順に操作しています。
であれば、__`Connect`の存在意義とは一体何なのか？__`Publish`して`Connect`するのなら、`Publish`内で自動的にやっておいてくれれば良いのではないか？と思うかもしれません。
ところが、`Connect`を呼び出すタイミングには重要な意味があって、場合によっては__`Connect`の呼び出すタイミングによって実行結果がまったく変わってくる__ことがあります。

例えば、以下のような例では、`Connect`を呼び出すタイミングが非常に重要になります。
①〜④のうち、どこで`Connect()`を呼び出すべきでしょうか？

```csharp
//1~10までの値を発行するColdなObservableソース
IObservable<int> observableRange = Observable.Range(1, 10);

IConnectableObservable<int> connectableObservable =
    observableRange.Publish();

//①ここでConnect()?
connectableObservable.Subscribe(n => Debug.Log($"1個めのSubscribe：{n}");
//②ここでConnect()?
connectableObservable.Subscribe(n => Debug.Log($"2個めのSubscribe：{n}"));
//③ここでConnect()?
connectableObservable.Subscribe(n => Debug.Log($"3個めのSubscribe：{n}"));
//④ここでConnect()?
```

↓
↓
↓
↓

答えは、3つの`Subscribe`にすべて値を通知させるためには、__④の位置__で`Connect()`を呼び出す必要があります。

`Connect()`の効果は「__`Publish`内の`Subject<T>`に、変換元のColdなIObservableを`Subscribe`させる__」でした。
この場合、「変換元のColdな`IObservable`」とは`Observable.Range`ですので、__`Subscribe`した瞬間に1~10までの値が即座に流れてきてしまいます__。

したがって、例えば①の位置で`Connect()`を呼び出すと、__誰も`Subject`を`Subscribe`しないまま1~10の値が流れてきてすぐに`OnCompleted`する__ので、結果として__`Observable.Range`が発行した値は虚無に消えます__。
`OnCompleted`したあとに3回`Subscribe`していますが、もはやなんの値も流れてこないので、値がまったく通知されてこないことになってしまいます。

②や③で`Connect()`した場合は、`Connect()`以前に`Subscribe`していた購読先には値が届きますが、`Connect()`以降に`Subscribe`した購読先には値は届きません。

④で`Connect()`すれば、きちんとすべての購読先が`Subscribe`を完了させ、準備万端な状態で`Observable.Range`を`Subscribe`するので、すべての購読先に対して`Observable.Range`の発行値をマルチキャストすることができます。

このように、ファクトリメソッドなどのColdなObservableソースをHot変換する場合は、`Connect()`を呼ぶタイミングが非常に重要になることが分かります。

###`Connect`のタイミングに左右されない場合

対して、`Connect()`を呼ぶタイミングが重要ではない場合もあります。それは、__ObservableソースがHot__である場合です。
ObservableソースがHotとは、__`Subject`から値が生成されている場合__を指します。

先ほどの例の`Observable.Range`ファクトリメソッドの場合は、__`Subscribe`したタイミングで、値が生成されて発行されていく__のでした。つまり、「__購読側で値の発行タイミングを制御できている__」と言えます。
これに対して`Subject`から値が生成されているということは、値の発行タイミングが`OnNext`が呼ばれるタイミングに委ねられることになります。つまり、「__購読側では値が発行されるタイミングを制御することができない__」、と言えます。

値の発行タイミングが制御できるのであれば、値の発行開始を示す`Connect()`呼び出しのタイミングは前述の通り非常に重要になりますが、`Subject`のように__値の発行タイミングが制御できないのであれば、もはや`Connect()`の呼び出すタイミングはいつでも良い__です。いつ`Subscribe`しようが、値が流れてくるタイミングはこちらでは制御できないためです。

例えば、次のような例で考えてみます。

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

    //中略
    //Subjectに対して何らかのタイミングでOnNextにより値を発行している

    //SubjectをIObservableとして公開している
    public IObservable<int> Observable => subject;
}

class SubjectSubscriber
{
    private SubjectProvider subjectProvider = new SubjectProvider();

    public SubjectSubscriber()
    {
        //外部が公開しているSubjectに対してオペレータをつけてHot変換
        IConnectableObservable<TimeSpan> connectableObservable =
            subjectProvider.Observable
                           .Where(n => n > 0)
                           .Select(n => TimeSpan.FromSeconds(n))
                           .Publish();
        //どこでConnectしよう？
        connectableObservable.Subscribe(time => Debug.Log(time));
        //どこでConnectしよう？
        connectableObservable.Subscribe(time => Debug.Log(time));
        //どこでConnectしよう？
        connectableObservable.Subscribe(time => Debug.Log(time));
        //どこでConnectしよう？
    }
}
```

`SubjectSubscriber`は、外部が公開している`Subject`を取得してHot変換をかけています。どこで`Connect`しようか迷っているようですが、この場合はどこで`Connect`しても結果は変わりません。
Observableソースが`Subject`の場合、いつ`Subscribe`しようが単一のObservableシーケンスを`Subscribe`することになるので、`Subscribe`のタイミングによって結果が左右されることがないのです。

ただ、もちろん`Subscribe`のタイミングが大幅に遅れると、それまで発行された値は通知されなくなるので、例えば以下のようにするのは問題です。

```csharp
IConnectableObservable<TimeSpan> connectableObservable =
    subjectProvider.Observable
                   .Where(n => n > 0)
                   .Select(n => TimeSpan.FromSeconds(n))
                   .Publish();

connectableObservable.Subscribe(time => Debug.Log(time));
connectableObservable.Subscribe(time => Debug.Log(time));
connectableObservable.Subscribe(time => Debug.Log(time));

//10秒経ってからソースSubjectをSubscribe
await Task.Delay(10000);
connectableObservable.Connect();
```

このようにすると、`connectableObservable`を`Subscribe`してから10秒間に発行された値は購読先にマルチキャストされなくなってしまうため、このような極端な例はNGとなります。

逆に、早く`Connect`する分にはまったく問題がありませんので、ObservableソースがHotであると分かっている場合は、`IConnectableObservable<T>`を即`Connect`してしまっても問題ないでしょう。

##`Connect`の戻り値の`IDisposable`とは何なのか

ところで、`Connect()`の戻り値は`IDisposable`となっています。果たしてこの`IDisposable`は、`Subscribe`の戻り値の`IDisposable`と何が違うのでしょうか？前に掲載した図をもう一度見てみると、違いはすぐに分かると思います。

![ConnectableObservable.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/4886bb5b-ea4d-2ad8-0dfa-2afeae1fd834.png)

この図を見ると、__`Connect`も`Subscribe`もどちらもやっていることは「`Subscribe`」__ですね。

したがって、`Connect`の戻り値の`IDisposable`とは、__`Subject`が変換元の`IObservable`を`Subscribe`したときに出てきた`IDisposable`のことである__ということがわかります。
`IConnectableObservable`の`Subscribe`の戻り値の`IDisposable`を`Dispose`すると、`Subject`への購読が`Dispose`されるのと同様に、__`Connect`の戻り値の`IDisposable`を`Dispose`すると、ソースとなるColdな`IObservable`への購読が`Dispose`される__ということになります。

###`Connect`または`Subscribe`を`Dispose`したときの挙動の違い

ではどちらを`Dispose`すべきなのか。

まず前提として、__ただ単に`Dispose`したいだけ、購読を解除したいだけならば、どちらを`Dispose`しても挙動は変わりません。__
`Subscribe`のほうを`Dispose`すれば`Subject`自体への購読が直接解除されるし、`Connect`のほうを`Dispose`すれば、`Subject`に入ってくるソースからの通知が遮断されるので、結果的に通知が来なくなります。

挙動が変わってくるのは、__`Connect`を`Dispose`したあとに、再び`Connect`（再接続）__するような場合です。そして、これに関しても、__ソースとなるColdな`IObservable`のObservableソースがColdなのかHotなのかで挙動が随分変わってきます__。

もし、__ObservableソースがHot__、つまり`Subject`から値が生成されているならば、何度「切断/再接続」を行ったとしても、ただ単に接続されている場合は値が飛んでくる、切断した場合は値が飛んでこない、それだけの違いになります。
しかし、__ObservableソースがCold__、つまりファクトリメソッドなどから値が生成されている場合は、__「切断/再接続」をするたびに、新しいObservableソースが生成される__ことになります。

例えば、以下の例を見てください。
`Observable.Interval`を`Publish`して`Connect`したあと、3秒後に切断/再接続を行う例です。

```csharp
IObservable<long> observableInterval = Observable.Interval(TimeSpan.FromSeconds(1));

IConnectableObservable<long> connectableObservable =
    observableInterval.Publish();

connectableObservable.Subscribe(n => Debug.Log($"1個めのSubscribe：{n}"));
connectableObservable.Subscribe(n => Debug.Log($"2個めのSubscribe：{n}"));

//Observale.Intervalに接続する
IDisposable connection = connectableObservable.Connect();

//3秒経ったら切断して再接続する
await Task.Delay(3000);
Debug.Log("3秒経ったので切断/再接続");
connection.Dispose();
connectableObservable.Connect();
```

実行結果は次のようになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/f1862899-3012-2f60-0d91-1a6fc2fc5fd2.png)

3秒経って「切断」を行うと、それまでのObservableシーケンスは破棄されます。そして、__「再接続」を行うと、また新しいObservableシーケンスが生成される__ため、インクリメントが0に戻っていることがわかります。

次の例のように、Observableソース自体をHot化しておけば、「切断/再接続」を行ってもObservableシーケンスは破棄されず、再接続しても途中から始まります。

```csharp

IConnectableObservable<long> connectableInterval =
    Observable.Interval(TimeSpan.FromSeconds(1))
              .Publish();
connectableInterval.Connect(); //Observable.Interval自体をHot化（ObservableソースがHot化する）

IConnectableObservable<long> connectableObservable =
    connectableInterval.Publish();

connectableObservable.Subscribe(n => Debug.Log($"1個めのSubscribe：{n}"));
connectableObservable.Subscribe(n => Debug.Log($"2個めのSubscribe：{n}"));

//Observale.IntervalをHot化したものに接続する
IDisposable connection = connectableObservable.Connect();

//3秒経ったら切断して再接続する
await Task.Delay(3000);
Debug.Log("3秒経ったので切断/再接続");
connection.Dispose();
connectableObservable.Connect();
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/697830/487f6357-5ad4-ddf4-0f9e-e02d879e57a1.png)

###挙動まとめ

`Connect`の`IDisposable`と`Subscribe`の`IDisposable`、それぞれの挙動をまとめると次のようになります。

| Obsevableソース | `Connect`を`Dispose` | `Subscribe`を`Dispose` |
| --- | --- | --- |
| __Hot__ | 通知が来なくなる<br>__再接続すると、そのタイミングからまた通知が再開される__ | 通知が来なくなる |
| __Cold__ | 通知が来なくなる<br>__再接続すると、新しいObservableシーケンスが生成されるので最初から値が発行される__ | 通知が来なくなる |

#まとめ

長くなってしまいましたがまとめると次のようになります。

- Hot変換とは、Observableシーケンスを`Subject`が代表して購読し、そこから再配信することである
- `Publish`オペレータはHot変換の準備をしているだけ。`Publish`をつけただけでは、Observableシーケンスになんの影響も与えない
- ObservableソースがHotかColdかによって`Connect`すべきタイミングが異なる
- 切断/再接続時の挙動もObservableソースのHotColdによって異なるので注意

自分の中で`Publish`の理解が曖昧だったので記事にまとめましたがだいぶしっかりした知識になった気がします。
あと基本、外部に`IObservable`を公開するときはHot変換して公開したほうが良さそうな気がします。外部からはこの`IObservable`がHotかColdか、はたまたObservableソースがColdなのかHotなのかなんてわかりませんから。外部に公開するときはHot、これを徹底しておけばハマる確率がかなり減らせるのではないかと思いました。もちろん意図的にColdで公開したい！って場合は、わかるようにしておけばそれはそれで良いと思いますけどね。

では、最後までありがとうございました！
ご意見ご指摘等ありましたらコメントお願いします！