如何在订阅者在 Rx 中的下一个回调之后执行操作？答案

【问题标题】：How to perform an action after the subscriber on next callback in Rx?如何在订阅者在 Rx 中的下一个回调之后执行操作？
【发布时间】：2013-01-07 19:49:32
【问题描述】：

我希望能够添加这样的扩展方法：

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

关于如何实现上述方法有什么建议吗？有没有我可以使用的内置方法？（假设：同步/立即订阅）

【问题讨论】：

你不能只订阅两次吗？

标签： system.reactive reactive-programming

【解决方案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);
    });
}

【讨论】：