更改RX运营商的间隔？

Question

这可能是一个愚蠢的问题,因为我对RX有点新:)

我正在抽样一个事件(RX为.Net 4.0):

eventAsObservable.Sample(TimeSpan.FromSeconds(1)).Timestamp().Subscribe(x => Console.WriteLine("testing:"+ x.Value.EventArgs.str));

问题是采样时间需要能够动态改变,我想我可以制作一些属性来移除现有的处理程序,并在它发生变化时创建一个新的处理程序,但它看起来有点乱,更容易受到计时问题的影响.有没有办法简单地改变间隔？

示例:假设某人正在编写一串字符,当检测到某个序列时,您希望更改采样时间而不会错过任何事件,并且最好不要多次获取事件

Answer 1

我不知道改变现有采样间隔的方法,但你可以做的是以你需要的最高频率采样,然后用一个Where使用你可以改变的变量的子句进行过滤.

例如:

static IObservable<T> SampleEvery<T>(this IObservable<T> source,
    Func<int> multipleProvider)
{
    int counter = 0;
    Func<T, bool> predicate = ignored => {
        counter++;
        if (counter >= multipleProvider())
        {
            counter = 0;
        }
        return counter == 0;
    };
    return source.Where(predicate);
}

然后你会这样称呼它:

// Keep this somewhere you can change it
int multiple = 1;

eventAsObservable.Sample(TimeSpan.FromSeconds(1))
                 .SampleEvery(() => multiple)
                 .Timestamp()
                 .Subscribe(x => Console.WriteLine("testing:" + 
                                                   x.Value.EventArgs.str));

现在,改变值multiple将改变有效采样频率.

这是一个非常难看的黑客,但我认为它应该工作.

Answer 2

我知道这个问题已经得到了回答,但我想我还会添加另外几种以Rx方式处理它的方法.

你可以使用Switch一系列TimeSpan的:

private Subject<TimeSpan> sampleFrequencies = new Subject<TimeSpan>();

sampleFrequencies
    .Select(x => eventAsObservable.Sample(Observable.Interval(x)).Timestamp())
    .Switch()
    .Subscribe(x => .WriteLine("testing:" + x.Value.EventArgs.str));

// To change:
// sampleFrequencies.OnNext(TimeSpan.FromSeconds(5));

或者,它也可以使用Defer,TakeUntil和Repeat(这个有点疯狂,并作为思想练习包括在内)解决:

private TimeSpan sampleFrequency = TiemSpan.FromSeconds(2);
private Subject<Unit> frequencyChanged = new Subject<Unit>();

(Observable
    .Defer(() => eventAsObservable
       .Sample(Observable.Interval(sampleFrequency)
    )
    .Timestamp()
    .TakeUntil(frequencyChanged)
).Repeat()
.Subscribe(x => .WriteLine("testing:" + x.Value.EventArgs.str));

// To change: 
// sampleFrequency = TimeSpan.FromSeconds(5);
// frequencyChanged.OnNext(new Unit());