Thêm một chuỗi quan sát được sau khi đăng ký

Question

Chúng tôi đang sử dụng Rx để theo dõi hoạt động trong ứng dụng Silverlight của chúng tôi để chúng tôi có thể hiển thị thông báo cho người dùng sau một khoảng thời gian không hoạt động.

Chúng tôi đang chuyển sự kiện (di chuyển chuột, v.v.) thành các quan sát và sau đó hợp nhất các quan sát lại với nhau để tạo ra một (allActivity) duy nhất có thể quan sát được. Sau đó chúng tôi điều chỉnh allActivity bằng cách sử dụng một khoảng thời gian và một thứ gì đó đăng ký để được thông báo khi hệ thống không hoạt động trong một khoảng thời gian.

Làm cách nào để thêm chuỗi/chuỗi quan sát mới vào sau đăng ký (để đăng ký chọn tùy chọn này mà không hủy đăng ký và đăng ký lại).

ví dụ: hợp nhất một vài chuỗi với nhau, điều tiết, đăng ký. Bây giờ thêm một chuỗi bổ sung vào quan sát đã được đăng ký.

Ví dụ mã:

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

Cách đơn giản nhất để làm điều này sẽ được sử dụng một đối tượng trung gian ở vị trí của Merge cuộc gọi.

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

Lớp Subject sẽ ngừng đi qua các sự kiện OnNext (và onerror xa hơn và OnCompleted) từ bất kỳ quan sát nó được đăng ký nếu nó nhận được bất kỳ onerror hoặc OnCompleted.

Sự khác biệt chính giữa cách tiếp cận này và mẫu của bạn là nó đăng ký tất cả các sự kiện khi chủ đề được tạo, thay vì khi bạn đăng ký với sự hợp nhất có thể quan sát được. Vì tất cả các quan sát trong ví dụ của bạn đều nóng, sự khác biệt không đáng chú ý.

Answer 2

Có một tình trạng quá tải để Merge mà mất trong một IObservable < IObservable <TSource> >. Đặt chuỗi bên ngoài thành Subject < IObservable <TSource> > và gọi OnNext đến đó khi bạn muốn thêm một nguồn khác vào nhóm. Nhà điều hành Hợp nhất sẽ nhận được nguồn và đăng ký với nó:

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