重置和处置 observable 订阅者，反应式扩展

Question

假设我有这个：

    public class UploadDicomSet 
{ 
    public UploadDicomSet()
    {
        var cachCleanTimer = Observable.Interval(TimeSpan.FromMinutes(2));
        cachCleanTimer.Subscribe(CheckUploadSetList);
        //Start subscriber
    }
    void CheckUploadSetList(long interval)
    {
        //Stop and dispose subscriber
    }
    public void AddDicomFile(SharedLib.DicomFile dicomFile)
    {
        //Renew subscriber, call CheckUploadSetList 2 minutes later
    }
}

1- in CheckUploadSetList 我要处理或完成 observable

2-在AddDicomFile我要重置它

作为方法中的注释。

更新：

我可以通过Timer 来做：

 public class UploadDicomSet : ImportBaseSet
{
    Timer _timer;
    public UploadDicomSet()
    {
        _timer = new Timer(CheckUploadSetList, null, 120000, Timeout.Infinite);
    }

    void CheckUploadSetList(object state)
    {
        Logging logging = new Logging(LogFile);
        try
        {
            _timer.Dispose(); //Stop the subscription
                              //dispose everything
        }
        catch (Exception exp)
        {
            logging.Log(ErrorCode.Error, "CheckUploadSetList() failed..., EXP:{0}", exp.ToString());
        }
    }
    public void AddDicomFile(SharedLib.DicomFile dicomFile)
    {
        _timer.Change(120000, Timeout.Infinite);
    }
}

提前致谢。

Answer 1

这种事情你应该使用Switch()。

类似这样的：

public class UploadDicomSet : ImportBaseSet
{
    IDisposable subscription;
    Subject<IObservable<long>> subject = new Subject<IObservable<long>>();

    public UploadDicomSet()
    {
        subscription = subject.Switch().Subscribe(s => CheckUploadSetList(s));
        subject.OnNext(Observable.Interval(TimeSpan.FromMinutes(2)));
    }

    void CheckUploadSetList(long interval)
    {
        subject.OnNext(Observable.Never<long>());
        // Do other things
    }

    public void AddDicomFile(SharedLib.DicomFile dicomFile)
    {
        subject.OnNext(Observable.Interval(TimeSpan.FromMinutes(2)));
        // Reset the subscription to go off in 2 minutes from now
        // Do other things
    }
}

Answer 2

将响应式扩展用于某些计时器功能对我来说似乎有点矫枉过正。为什么不为此使用普通计时器，并在给定时间启动/停止它？

让我给出一个想法。

public class UploadDicomSet : ImportBaseSet
{
    IDisposable subscription;

    public void CreateSubscription()
    {
        var cachCleanTimer = Observable.Interval(TimeSpan.FromMinutes(2));

        if(subscription != null)
            subscription.Dispose();

        subscription = cachCleanTimer.Subscribe(s => CheckUploadSetList(s));
    }

    public UploadDicomSet()
    {
        CreateSubscription();
        // Do other things
    }

    void CheckUploadSetList(long interval)
    {
        subscription.Dispose(); // Stop the subscription
        // Do other things
    }

    public void AddDicomFile(SharedLib.DicomFile dicomFile)
    {
        CreateSubscription(); // Reset the subscription to go off in 2 minutes from now
        // Do other things
    }
}

背景材料

我真的可以推荐这些网站：

http://www.introtorx.com/

http://rxwiki.wikidot.com/101samples