更改 RX 运算符的间隔？

Question

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

我正在采样一个事件（.Net 4.0 的 RX）：

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 的值将更改有效采样频率。

这是一个非常丑陋的 hack，但我认为它应该可以工作。

Answer 2

我知道这个问题已经得到解答，但我想我会添加另外几种以 Rx 方式解决它的方法。

您可以在TimeSpan 的序列上使用Switch：

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());

Answer 3

TL;DR: 使用 ObservableFromIntervalFunctor 创建一个 Observable，如下所示：

void Main()
{
    // Pick an initial period, it can be changed later.
    var intervalPeriod = TimeSpan.FromSeconds(1);

    // Create an observable using a functor that captures the interval period.
    var o = ObservableFromIntervalFunctor(() => intervalPeriod);

    // Log every value so we can visualize the observable.
    o.Subscribe(Console.WriteLine);

    // Sleep for a while so you can observe the observable.
    Thread.Sleep(TimeSpan.FromSeconds(5.0));

    // Changing the interval period will takes effect on next tick.
    intervalPeriod = TimeSpan.FromSeconds(0.3);

}

IObservable<long> ObservableFromIntervalFunctor(Func<TimeSpan> intervalPeriodFunctor)
{
    return Observable.Generate(0L, s => true, s => s + 1, s => s, s => intervalPeriodFunctor());
}

解释： Observable.Generate 有一个重载，允许您指定通过仿函数生成下一个值的时间。通过传递一个捕获时间跨度变量的函子，您可以通过更改捕获的时间跨度变量来改变 observable.interval 周期。

Linqpad sn-p here

Answer 4

你为什么不订阅两次？

Observable.Merge(
    eventAsObservable.Sample(TimeSpan.FromSeconds(1)).Timestamp().SelectMany(x => doLocalLookup(x)),
    eventAsObservable.Sample(TimeSpan.FromSeconds(10)).Timestamp().SelectMany(x => doRemoteLookup(x)),
).Subscribe(Console.WriteLine);

或者，如果搜索仅基于某种前缀或限定符（如 Google Chrome 的“？”）处于活动状态运营商：

Observable.Merge(
    eventAsObservable.Sample(TimeSpan.FromSeconds(1)).Where(x => isLocal(x)).SelectMany(x => doLocalLookup(x)),
    eventAsObservable.Sample(TimeSpan.FromSeconds(10)).Where(x => isARemoteQuery(x).SelectMany(x => doRemoteLookup(x)),
).Subscribe(Console.WriteLine);