はじめに

Reactive Extension(C#)で、Observable同士の合成の方法で少してこずったのでメモします。

やりたかった事

あるObservable(A)が返した値を次の引数を受け取るObservable(B)に渡して、(A)と(B)を一つのObservableとして扱うという事を実現したかった。
具体的に言うとAccess Tokenを返すObservableからAccess Tokenを取得し、それを引数としてUser Nameを返すObservableに渡す、というような流れを一つに合成したいと思いました。

合成なしの例

データ取得する上で以下のようなRepositoryがあるとします。

'''

    public class AuthRepository 
    {
        private readonly static string _accessToken = "your access token";

        public IObservable<string> GetAccessTokenObservable()
        {
            // AccessTokenを発行
            return Observable.Return(_accessToken);
        }

        public IObservable<string> GetUserNameObservable(string accessToken)
        {
            if (accessToken == _accessToken)
            {
                // 発行したAccessTokenと同じ値ならUserNameを返す
                return Observable.Return("your name");
            }
            else 
            {
                return Observable.Return("anonymous");
            }
        }
    }

'''

GetAccessTokenObservable()でAccess Tokenを取得します。
GetUserNameObservable(string)にAccess Tokenを渡してUser Nameを取得します。
一連の処理を合成なしにひとまとめに書いたら以下のようになると思います。

'''

        var repository = new AuthRepository();
        repository.GetAccessTokenObservable()
                  .Subscribe(x =>
                  {
                      repository.GetUserNameObservable(x).Subscribe(y =>
                      {
                          Console.WriteLine("User Name:" + y); // 適当に出力
                      });
                  });

'''

これでも一応AccessTokenを取得して、それを利用してUserNameを取得するという目的は達成できるのですが、OnErrorなどの処理を加えようとすると二重に記述することとなり面倒です。
Subscribeを二重に行う事によりIDisposableが二重に発行されてしまうのも管理が面倒となるポイントだと思います。

合成

結論から言うと以下のやり方でIObservableの一本化ができました。

'''

        repository.GetAccessTokenObservable()
                  .Select(x => repository.GetUserNameObservable(x))
                  .Merge()
                  .Subscribe(x => Console.WriteLine("User Name:" + x));

'''
流れとしては

GetAccessTokenObservable()でAccessTokenを取得
Select()でGetUserNameObservable()に値を渡しIObservable<IObservable<string>>に変換
Merge()でIObservable<string>に変換
Subscribe()で購読

となってます。
Select()を行った時点では値はIObservable<IObservable<string>>となってますので、その後Merge()でIObservable<string>に変換する必要がある、という所がちょっとつまずいたポイントでした。

参考
Reactive Extensions再入門その４０「IObservableの合成はじめました」https://blog.okazuki.jp/entry/20120218/1329581391