处理订阅时访问当前窗口答案

【问题标题】：Accessing current window when disposing of subscription处理订阅时访问当前窗口
【发布时间】：2012-06-27 09:44:30
【问题描述】：

我有以下代码：

var observable = ... subscribe to event here ...

var windows = observable.Window(TimeSpan.FromSeconds(240));

aggregatedWindows = windows.SelectMany(
    window => window.Aggregate(new Context(), AggregateContext));

subscription = aggregatedWindows.Subscribe(OnWindow);

... later

subscription.Dispose();

想象一个场景，当我正在处理一个窗口并且有人请求我的应用程序应该关闭时。我将处理此订阅，这将停止正在处理的事件，但我也会丢失最后一个信息窗口。

我不确定解决这个问题的最佳方法是什么......

我可以在最后一次看到的窗口通过聚合函数时存储本地状态（但这似乎是错误的）...

任何帮助将不胜感激！

【问题讨论】：

标签： c# system.reactive

【解决方案1】：

您可以对窗口进行操作，而不是保留聚合的订阅 - 这是您主要希望保持连接以使最后一个窗口到达的内容，并使用超时断开连接以防分区时间过长.

这里使用了一个单独的类，因为使用Create 会使其自动分离 - 在进行 dispose 调用后会立即断开观察者的连接。所以从根本上说，Dispose 的含义就是这里发生了变化。

    public static IObservable<T> DeferDisconnection<T>(this IObservable<T> observable, TimeSpan timeout)
    {
        return new ClosingObservable<T>(observable, timeout);
    }


    public class ClosingObservable<T> : IObservable<T>
    {

        private readonly IConnectableObservable<T> Source;
        private readonly IDisposable Subscription;
        private readonly TimeSpan Timeout;

        public ClosingObservable(IObservable<T> observable, TimeSpan timeout)
        {
            Timeout = timeout;
            Source = observable.Publish();
            Subscription = Source.Connect();
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            Source.Subscribe(observer);

            return Disposable.Create(() => Source.Select(_ => new Unit())
                                                 .Amb(Observable.Timer(Timeout).Select(_ => new Unit()))
                                                 .Subscribe(_ => Subscription.Dispose())
                                                 );
        }
    }

测试：

            var disposable =
            Observable.Interval(TimeSpan.FromSeconds(2))
                      .Do(Console.WriteLine)
                      .DeferDisconnection(TimeSpan.FromSeconds(5))
                      .Subscribe();

            Console.ReadLine();

            disposable.Dispose();

            Console.ReadLine();

【讨论】：

我很困惑这是如何解决问题的。看起来 DeferDisconnection 会在处理之前稍等片刻，但如果窗口很大，那么这将导致应用程序等待很长时间才能关闭。对吗？
@jonnii 这就是为什么有一个 timeout 参数 - 它会在下一个值到达时断开连接，或者如果超时参数太长，它就会断开连接。

【解决方案2】：

这可以通过最后显示的部分窗口来确认。

class Program
{
    public class Context
    {
        public int count;
    }

    static Context AggregateContext(Context c, long i)
    {
        c.count++;
        return c;
    }

    static void OnWindow(Context c) { Console.WriteLine(c.count); }

    static void Main(string[] args)
    {
        var canceled = new Subject<bool>();

        var observable = Observable.Interval(TimeSpan.FromSeconds(.1)).TakeUntil(canceled);

        var windows = observable.Window(TimeSpan.FromSeconds(3));

        var aggregatedWindows = windows.SelectMany(
            window => window.Aggregate(new Context(), AggregateContext));

        var subscription = aggregatedWindows.Subscribe(OnWindow);

        Thread.Sleep(TimeSpan.FromSeconds(10));

        canceled.OnNext(true);
        subscription.Dispose();

        Console.WriteLine( @"Output should have been something like 30,30,30,30,10" );
        Console.ReadLine();
    }
}

【讨论】：