在响应式扩展中创建优先级流并忽略第二个流答案

【问题标题】：Creating a priority stream in reactive extensions and ignoring second stream在响应式扩展中创建优先级流并忽略第二个流
【发布时间】：2013-11-08 01:50:47
【问题描述】：

我有两个可观察对象，它们各自做不同的事情。如果他们的第一个流在某个时间间隔内接收到一些数据，我希望第二个流忽略它的输入，比如说 500 毫秒。我该怎么做呢？

var s1 = new Subject<string>();
var s2 = new Subject<string>();

s1.Subscribe(x => Debug.Write(x));                
//Subscribe something slightly different
s2.Subscribe(x => Debug.WriteLine(x));


s1.OnNext("foo");
Thread.Sleep(500);
s2.OnNext("bar");

/* Expected output:
    * foobar /r/n
    */

s1.OnNext("fizz");
s2.OnNext("buzz");
/* Expected output:
* fizz
*/

谢谢！

【问题讨论】：

标签： system.reactive

【解决方案1】：

这将在每次s1 产生一个值时启动一个计时器。每当该计时器运行时，它将取消订阅s2，并在计时器到期时重新订阅：

var delay = TimeSpan.FromMilliseconds(500);
var s2lowPriority = s1
    .Select(_ => Observable.Timer(delay).SelectMany(_ => s2))
    .StartWith(s2)
    .Switch();

s2lowPriority.Subscribe(x => Debug.WriteLine(x));

【讨论】：

这是一个很好的解决方案。