c# - 订阅后添加可观察序列

Question

我们正在使用 Rx 来监视我们的 silverlight 应用程序中的事件，以便我们可以在一段时间不活动后向用户显示一条消息。

我们正在将事件(鼠标移动等)转化为可观察对象，然后将这些可观察对象合并在一起以创建一个 (allActivity) 可观察对象。然后，我们使用时间跨度限制 allActivity 可观察对象，并在系统处于非事件状态一段时间后订阅通知。

我如何在订阅后向它添加一个新的可观察/序列(以便订阅在不取消订阅和重新订阅的情况下选择它)。

例如将几个序列合并在一起，节流，订阅。现在向已订阅的可观察对象添加一个额外的序列。

示例代码:

private IObservable<DateTime> allActivity;
public void CreateActivityObservables(UIElement uiElement)
{
    // Create IObservables of event types we are interested in and project them as DateTimes
    // These are our observables sequences that can push data to subscribers/ observers 
    // NB: These are like IQueryables in the sense that they do not iterate over the sequence just provide an IObservable type
    var mouseMoveActivity = Observable.FromEventPattern<MouseEventHandler, MouseEventArgs>(h => uiElement.MouseMove += h, h => uiElement.MouseMove -= h)
                                      .Select(o => DateTime.Now);

    var mouseLeftButtonActivity = Observable.FromEventPattern<MouseButtonEventHandler, MouseButtonEventArgs>(h => uiElement.MouseLeftButtonDown += h, h => uiElement.MouseLeftButtonDown -= h)
                                            .Select(o => DateTime.Now);

    var mouseRightButtonActivity = Observable.FromEventPattern<MouseButtonEventHandler, MouseButtonEventArgs>(h => uiElement.MouseRightButtonDown += h, h => uiElement.MouseRightButtonDown -= h)
                                             .Select(o => DateTime.Now);

    var mouseWheelActivity = Observable.FromEventPattern<MouseWheelEventHandler, MouseWheelEventArgs>(h => uiElement.MouseWheel += h, h => uiElement.MouseWheel -= h)
                                       .Select(o => DateTime.Now);

    var keyboardActivity = Observable.FromEventPattern<KeyEventHandler, KeyEventArgs>(h => uiElement.KeyDown += h, h => uiElement.KeyDown -= h)
                                     .Select(o => DateTime.Now);

    var streetViewContainer = HtmlPage.Document.GetElementById("streetViewContainer");
        var mouseMoveHandler = new EventHandler<HtmlEventArgs>(this.Moo);
        bool b = streetViewContainer.AttachEvent("mousemove", mouseMoveHandler);

    var browserActivity = Observable.FromEventPattern<Landmark.QDesk.ApplicationServices.IdleTimeoutService.MouseMoveHandler, HtmlEventArgs>(h => this.MyMouseMove += h, h => this.MyMouseMove -= h).Select(o => DateTime.Now);

    // Merge the IObservables<DateTime> together into one stream/ sequence
    this.allActivity = mouseMoveActivity.Merge(mouseLeftButtonActivity)
                                        .Merge(mouseRightButtonActivity)
                                        .Merge(mouseWheelActivity)
                                        .Merge(keyboardActivity)
                                        .Merge(browserActivity);
}

public IDisposable Subscribe(TimeSpan timeSpan, Action<DateTime> timeoutAction)
{
    IObservable<DateTime> timeoutNotification = this.allActivity.Merge   (IdleTimeoutService.GetDateTimeNowObservable())
                                                                .Throttle(timeSpan)
                                                                    .ObserveOn(Scheduler.ThreadPool);

    return timeoutNotification.Subscribe(timeoutAction);
}

Answer 1

Merge 有一个重载，它接收一个 IObservable >。使外部序列成为 Subject > 并在您想要向该束中添加另一个源时调用它的 OnNext。 Merge 运算符将接收源并订阅它:

var xss = new Subject<IObservable<int>>();
xss.Merge().Subscribe(x => Console.WriteLine(x));

xss.OnNext(Observable.Interval(TimeSpan.FromSeconds(1.0)).Select(x => 23 + 8 * (int)x));
xss.OnNext(Observable.Interval(TimeSpan.FromSeconds(0.8)).Select(x => 17 + 3 * (int)x));
xss.OnNext(Observable.Interval(TimeSpan.FromSeconds(1.3)).Select(x => 31 + 2 * (int)x));
...