订阅作为提出的最后一个方法答案

【问题标题】：subscribe as last method raised订阅作为提出的最后一个方法
【发布时间】：2014-08-19 09:48:15
【问题描述】：

有没有办法订阅一个方法，尽管它会在引发 onNext 时最后调用？

m_subject.Subscribe(() => Console.writeLine("firstSubscription");
m_subject.SubscribeLast(() => Console.writeLine("secondSubscription");
m_subject.Subscribe(() => Console.writeLine("thirdSubscription");

m_subject.OnNext();

// prints:
// firstSubscription
// thirdSubscription
// secondSubscription

【问题讨论】：

您能进一步解释一下吗？不太确定你想做什么！
我有一个处理资源的方法。必须在调用所有注册方法并使用资源后最后调用它。不知道订阅顺序
你放的所有代码，是在一个方法上吗？
我会有一个帮助类（ResourceManager 或其他东西）来包装资源并管理资源的订阅者。当没有更多订阅者时，它会释放资源。然后让您的 RX 查询向资源注册自己，并在完成后取消注册。
一般来说，我建议 Observable 序列 (IObservable<T>) 的值应该是不可变的数据类型。这对我来说意味着他们不应该有像资源处理这样的变异/副作用行为。也许您想查看不同的设计，或者扩展您的问题的具体细节，以便提供更好的指导

标签： c# system.reactive

【解决方案1】：

您不能让订阅者最后执行，但您可以将所有调用封装在一个订阅中。

类似的东西：

Action action = () => {};
Action lastAction = () => {};

m_subject.Subscribe(() => 
{
    action();
    lastAction();
});

action += (() => Console.writeLine("firstSubscription");
lastAction += (() => Console.writeLine("secondSubscription");
action += (() => Console.writeLine("thirdSubscription");

m_subject.OnNext();

// prints:
// firstSubscription
// thirdSubscription
// secondSubscription

【讨论】：

【解决方案2】：

您也可以通过定义一个自定义Subject<T> 来做到这一点，该Subject<T> 内部有一个默认主题和最后一个主题。

更新

我添加了ObserveOn 的重载来存储IScheduler 和SynchronizationContext，然后在订阅时应用它们。可以使用类似的技术使SubscribeOn 也能正常工作。

public class SubscribeLastSubject<T> : ISubject<T>, IDisposable
{
    private readonly Subject<T> subject = new Subject<T>();
    private readonly Subject<T> lastSubject = new Subject<T>();
    private IScheduler observeScheduler;
    private SynchronizationContext observerContext;

    public void OnNext(T value)
    {
        subject.OnNext(value);
        lastSubject.OnNext(value);
    }

    public void OnError(Exception error)
    {
        subject.OnError(error);
        lastSubject.OnError(error);
    }

    public void OnCompleted()
    {
        subject.OnCompleted();
        lastSubject.OnCompleted();
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return GetObservable().Subscribe(observer);           
    }

    public IDisposable SubscribeLast(IObserver<T> observer)
    {
        return GetLastObservable().Subscribe(observer);     
    }

    public IDisposable SubscribeLast(Action<T> action)
    {
        return GetLastObservable().Subscribe(action);
    }

    public SubscribeLastSubject<T> ObserveOn(IScheduler scheduler)
    {
        observeScheduler = scheduler;
        return this;
    }

    public SubscribeLastSubject<T> ObserveOn(SynchronizationContext context)
    {
        observerContext = context;
        return this;
    }

    public void Dispose()
    {
        subject.Dispose();
        lastSubject.Dispose();
    }

    private IObservable<T> GetObservable()
    {
        if (observerContext != null)
        {
            return subject.ObserveOn(observerContext);
        }

        if (observeScheduler != null)
        {
            return subject.ObserveOn(observeScheduler);
        }

        return subject;
    }

    private IObservable<T> GetLastObservable()
    {
        if (observerContext != null)
        {
            return lastSubject.ObserveOn(observerContext);
        }

        if (observeScheduler != null)
        {
            return lastSubject.ObserveOn(observeScheduler);
        }

        return lastSubject;
    }
}

用法

var m_subject = new SubscribeLastSubject<string>();

m_subject.ObserveOn(Scheduler.CurrentThread).Subscribe(s => Console.WriteLine("firstSubscription"));
m_subject.ObserveOn(Scheduler.CurrentThread).SubscribeLast(s => Console.WriteLine("secondSubscription"));
m_subject.ObserveOn(Scheduler.CurrentThread).Subscribe(s => Console.WriteLine("thirdSubscription"));

m_subject.OnNext("1");

Console.ReadKey();

输出

firstSubscription
thirdSubscription
secondSubscription

【讨论】：

您能解释一下这些具体有什么帮助吗？
这个解决方案很好，但我不能使用 ObserveOn。你有解决办法吗：m_subject.ObserveOn(...).SubscribeLast(...);