订阅后添加可观察序列

Question

我们使用Rx监控我们的Silverlight应用程序中的活动，以便在闲置一段时间后向用户显示消息。

我们将事件（鼠标移动等）转变为观察对象，然后将观察对象合并在一起以创建单个（allActivity）可观察对象。然后，我们使用一个时间间隔来限制allActivity可观察值，并且当系统在一段时间内处于非活动状态时，会预订通知。

如何在订阅后为此添加新的observable/sequence（以便订阅在未取消订阅和重新订阅的情况下选择此项）。

例如合并几个序列，油门，订阅。现在向已订阅的观测值添加一个附加序列。

示例代码：

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呼叫的中间主体的最简单方法。

Subject<DateTime> allActivities = new Subject<DateTime>(); 
var activitySubscriptions = new CompositeDisposable(); 

activitySubscriptions.Add(mouseMoveActivity.Subscribe(allActivities)); 
activitySubscriptions.Add(mouseLeftButtonActivity.Subscribe(allActivities)); 
//etc ... 

//subscribe to activities 
allActivities.Throttle(timeSpan) 
      .Subscribe(timeoutAction); 

//later add another 
activitySubscriptions.Add(newActivity.Subscribe(allActivities)); 

的Subject类将停止传递OnNext（和进一步的OnError和OnCompleted）事件从任何其订阅了，如果它接收到的任何的OnError或OnCompleted的可观察量。

此方法与您的示例之间的主要区别在于，它订阅创建主题时的所有事件，而不是订阅合并的observable时。由于您示例中的所有观测值都很热，所以差异不应该引人注意。

Answer 2

合并需要IObservable超负荷< IObservable <TSource> >。使外部序列成为主题< IObservable <TSource> >并在您想向其中添加其他源时调用OnNext。合并运营商将收到来源并订阅它：

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