system.reactive - Rx の次のコールバックでサブスクライバーの後にアクションを実行する方法は?

Question

次のような拡張メソッドを追加できるようにしたいと思います。

IObservable<T> AfterSubscriber(this IObservable<T> observable, Action<T> action)

どこ：

var subj = new Subject<int>();
subj.
    AfterSubscriber(i => Console.WriteLine("Second")).
    Subscribe(i => Console.WriteLine("First"));

subj.OnNext(1);

出力を生成します。

First
Second

上記の方法を実装する方法に関する提案はありますか? 使用できる組み込みメソッドはありますか? (前提: 同期/即時サブスクリプション)

Answer 1

この実装はどうですか？

public static IObservable<T> AfterSubscriber<T>(
    this IObservable<T> @this, Action<T> after)
{
    if (@this == null) throw new ArgumentNullException("@this");
    if (after == null) throw new ArgumentNullException("after");
    return Observable.Create<T>(o =>
    {
        return @this.Subscribe(t =>
        {
            o.OnNext(t);
            try
            {
                after(t);
            }
            catch (Exception exception)
            {
                o.OnError(exception);
            }
        }, o.OnError, o.OnCompleted);
    });
}