訂閱後添加可觀察序列

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