サンプルのまとめ

　今回のサンプルより、ObservableとObserverのデータ通知の関係をまとめると次のようになります。

• Observable内のsubscriber.onNext(item)にデータitemを渡すと、ObserverのonNext(T item)が実行され、引数に通知されたデータが渡ってくる。
• Observable内のsubscriber.onCompleted()を呼ぶと、ObserverのonCompleted()が実行される。
• Observable内でエラーが発生した場合、ObserverのonError(throwable error)が実行され、引数に発生したエラーオブジェクトが渡ってくる。

　主なポイントはこのようになりますが、今回のようにcreateメソッドを使ってObservableを生成する場合は、次のことも考慮しなくてはいけません。

• 完了もしくはエラーを通知した後はすみやかに処理を終える。
• 購読解除時の処理を入れる。

　また、Observableの処理に終わりがある場合は、onCompletedメソッドを呼んで完了の通知を行うことを忘れないようにしないといけません。

サンプルの簡略化

　今回のサンプルは最初のサンプルということで、どのようにObservableがデータを生成し通知するのかを見せるために、Observableのcreateメソッドを使って、データの生成から通知の処理を直接実装しました。そのため、購読解除時の処理や完了の通知などを実装者が気を付けて実装しないと適切な処理が行われない可能性があります。しかし、Observableの生成メソッドにはcreateメソッド以外にも、いろいろなメソッドが用意されており、すでに通知するデータが決まっている場合はObservableのjustメソッドのように、通知するデータを設定するだけで、それらのデータを通知するObservableを生成するようなメソッドも用意されています。

　また、subscribeメソッドの引数にはObserverの代わりに、各処理ごとの関数型インタフェースを渡すことも可能です。特に完了の通知やエラーの通知時の処理が不要な場合は、完了やエラーの通知を省略してデータ通知時の処理のみを受け取るsubscribeメソッドも用意されています。

　これらのことを踏まえて、最初のサンプルとほぼ同じことをする次のサンプルを作ってみましょう。まず、Observableのjustメソッドを使ってあいさつの言葉を通知するObservableを生成しています。次にObserverの代わりに各通知時の処理を行う関数型インタフェースを渡して通知を受け取った処理を実装しています。

// あいさつの言葉を標準出力するサンプル
public static void main(String[] args) throws Exception {
  
  // あいさつの言葉を通知するObservableの生成
  Observable<String> observableGreeting =
      Observable.just("Hello, World!", "こんにちは、世界！");
  
  // Observableを購読し処理を開始する
  observableGreeting.subscribe(
      // Observableからのデータを受け取った際の処理
      item -> {
        // 実行しているThread名の取得
        String threadName = Thread.currentThread().getName();
        // Observableからのデータをそのまま標準出力する
        System.out.println(threadName + ": " + item);
      },
      // Observableからエラーを通知された際の処理
      error -> {
        error.printStackTrace();
      },
      // Observableから完了を通知された際の処理
      () -> {
        // 実行しているThread名の取得
        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + ": 完了しました");
      });
}

　ちなみに最初のサンプルとの処理の違いは、justメソッドで生成されたObservableの場合、各データを通知するたびに購読解除の確認をしていることです。そのため、このサンプルだと仮に"Hello, World!"を通知した後に購読解除された場合、"こんにちは、世界！"は通知されません。それでは、このサンプルの実装について見ていきましょう。

　まず、Observableの生成ではObservableのjustメソッドが使われています。このjustメソッドは引数に渡したデータを順に通知するObesrvableを生成するメソッドです。このjustメソッドを使えば、通知するデータを引数に渡すだけで、購読解除時の対応も完了の通知も実装されているObservableが生成されます。

　次の変更箇所はsubscribeメソッドの引数にObserverではなく関数型インタフェースを渡しているところです。subscribeメソッドにはObserverを受け取るものの他に各通知の処理を行う関数型インタフェースを受け取るものがあり、中には完了時の処理やエラー時の処理が省略されているメソッドもあります。仮に今回のサンプルで完了時とエラー時の処理を行わず、データを受け取った時の処理のみを行うようにする場合は次のようにすることができます。

// Observableを購読し処理を開始する
observableGreeting.subscribe(
    // Observableからのデータを受け取った際の処理
    item -> {
      // 実行しているThread名の取得
      String threadName = Thread.currentThread().getName();
      // Observableからのデータをそのまま標準出力する
      System.out.println(threadName + ": " + item);
    });