ReactiveXを説明できるようになるシリーズその2 - Subscribe

前回はこちら

今回は、Subscribeについて見てみたいと思います。

Subscribeってなに？

Subscribe自体は、言葉の通り「購読する」みたいな意味で、Observableの中でemitされたアイテムを受信する受信者みたいな捉え方でいいのかなと思います。

The Observable Contractを見ると、こちらには以下のように書いてあります。

An observer communicates with its Observable by means of the following notifications:

Subscribe
indicates that the observer is ready to receive notifications from the Observable

「Subscribeは、ObserverがObservableからの通知を受信する準備ができてることを指してるよ」みたいに書かれています。

事実、RxJavaでSubscriberの実装を見ると、

Subscriber.java

public abstract class Subscriber<T> implements Observer<T>, Subscription {
...

といったようにObserverをimplementしているのがわかるので、やはり受信者という認識で問題無さそうです。

ちなみに、Subscriber自体は抽象クラスで、Observerインターフェースのメソッドについては何も触ってないので、subscribeするときにonNextとかを実装してやらないといけないんですね、なるほど...

Subscribeするとどうなるのか

Subscribeすると何が帰ってくるのかは、Javadocを見ると分かるように、Subscriptionが帰ってきます。
Subscription自体はunsubscribeするためのインターフェースですので、一度subscribeするとあとはそれを止めることしかできないって訳ですね。

じゃあ実際に、subscribeされた時に中では何をしてるのか見てみます。
https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/Observable.java#L8254
メソッド的にはここですが、実際この中でsubscribeっぽい処理をしているのは下のあたりだと思います。

Observable.java

try {
	// allow the hook to intercept and/or decorate
    hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
    return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
...

このうち、hook(RxJavaObservableExecutionHook)のonSubscribeStartとonSubscribeReturnについては、現状単に引数を返しているだけなので、実体としてはOnSubscribe.call(subscriber)って事になりそうです。
OnSubscribeについては、Observable生成時に必須のパラメータとなるので、ようはSubscriberを引数にOnSubscribeを呼んでるだけっぽいですね。

つまり

Subscribeについては、「Observableからの通知を受け取る」という意味で、Subscriberについてはその受信するためのクラスという風に言えそうです。
また、ObservableをSubscribeする事については「Observableの中のOnSubscribeがSubscriberを引数にcallされ、Observableの中でemitされると、Subscriberにそれが送られる」という事になりそうです。

実際は、OnSubscribeがcallされるといってもHot ObservableとCold Observableでニュアンスが変わるような気もするのですが、それは別の機会に見てみたいと思います。