Z rozszerzeniami reaktywnymi (RX), czy możliwe jest dodanie polecenia "Przerwa"?

Question

Mam klasę, która bierze strumień wydarzeń i wypycha inny strumień wydarzeń.

Wszystkie zdarzenia używają rozszerzeń reaktywnych (RX). Przychodzący strumień zdarzeń jest wypychany z zewnętrznego źródła do postaci IObserver<T> przy użyciu .OnNext, a wychodzący strumień zdarzeń jest wypychany przy użyciu IObservable<T> i .Subscribe. Używam Subject<T> do zarządzania tym, za kulisami.

Zastanawiam się, jakie techniki istnieją w RX, aby tymczasowo wstrzymać wyjście. Oznaczałoby to, że zdarzenia przychodzące narastałyby w wewnętrznej kolejce, a po ich wstrzymaniu zdarzenia miałyby być ponownie wyprowadzane.

Answer 1

Możesz symulować wstrzymywanie/anulowanie z Observable.

Gdy pauseObservable wyemituje "wstrzymaną" wartość, zdarzenia bufora do momentu, aż pauseObservable wyemituje wartość "wstrzymaną".

Oto przykład, który korzysta BufferUntil implementation by Dave Sexton i Observable logic by Timothy Shields (z własnego pytanie jakiś czas temu)

 //Input events, hot observable 
     var source = Observable.Interval(TimeSpan.FromSeconds(1)) 
      .Select(i => i.ToString()) 
      .Publish().RefCount(); 

     //Simulate pausing from Keyboard, not actually relevant within this answer 
     var pauseObservable = Observable.FromEventPattern<KeyPressEventHandler, KeyPressEventArgs>(
      k => KeyPressed += k, k => KeyPressed -= k) 
      .Select(i => i.EventArgs.PressedKey) 
      .Select(i => i == ConsoleKey.Spacebar) //space is pause, others are unpause 
      .DistinctUntilChanged(); 

     //Let events through when not paused 
     var notPausedEvents = source.Zip(pauseObservable.MostRecent(false), (value, paused) => new {value, paused}) 
      .Where(i => !i.paused) //not paused 
      .Select(i => i.value) 
      .Subscribe(Console.WriteLine); 

     //When paused, buffer until not paused 
     var pausedEvents = pauseObservable.Where(i => i) 
      .Subscribe(_ => 
       source.BufferUntil(pauseObservable.Where(i => !i)) 
        .Select(i => String.Join(Environment.NewLine, i)) 
        .Subscribe(Console.WriteLine)); 

Pokój dla poprawy: może scalić dwie subskrypcje do źródła (pausedEvents i notPausedEvents) jako jeden.

Answer 2

Oto rozsądnie prosty sposób Rx, aby zrobić to, co chcesz. Stworzyłem metodę rozszerzenia o nazwie Pausable, która ma źródło obserwowalne, a druga obserwowalną z boolean, która zatrzymuje lub wznawia obserwowalne.

public static IObservable<T> Pausable<T>(
    this IObservable<T> source, 
    IObservable<bool> pauser) 
{ 
    return Observable.Create<T>(o => 
    { 
     var paused = new SerialDisposable(); 
     var subscription = Observable.Publish(source, ps => 
     { 
      var values = new ReplaySubject<T>(); 
      Func<bool, IObservable<T>> switcher = b => 
      { 
       if (b) 
       { 
        values.Dispose(); 
        values = new ReplaySubject<T>(); 
        paused.Disposable = ps.Subscribe(values); 
        return Observable.Empty<T>(); 
       } 
       else 
       { 
        return values.Concat(ps); 
       } 
      }; 

      return pauser.StartWith(false).DistinctUntilChanged() 
       .Select(p => switcher(p)) 
       .Switch(); 
     }).Subscribe(o); 
     return new CompositeDisposable(subscription, paused); 
    }); 
} 

Może być stosowany tak:

var xs = Observable.Generate(
    0, 
    x => x < 100, 
    x => x + 1, 
    x => x, 
    x => TimeSpan.FromSeconds(0.1)); 

var bs = new Subject<bool>(); 

var pxs = xs.Pausable(bs); 

pxs.Subscribe(x => { /* Do stuff */ }); 

Thread.Sleep(500); 
bs.OnNext(true); 
Thread.Sleep(5000); 
bs.OnNext(false); 
Thread.Sleep(500); 
bs.OnNext(true); 
Thread.Sleep(5000); 
bs.OnNext(false); 

Teraz jedyną rzeczą, nie mogłem się zorientować, co masz na myśli przez „przychodzącego strumienia zdarzeń jest IObserver<T>”. Strumienie to IObservable<T>. Obserwatorzy nie są strumieniami. Wygląda na to, że czegoś tu nie robisz. Czy możesz dodać swoje pytanie i wyjaśnić dalej?

Answer 3

Oto moje rozwiązanie z użyciem operatorów buforze i okno:

public static IObservable<T> Pausable<T>(this IObservable<T> source, IObservable<bool> pauser) 
{ 
    var queue = source.Buffer(pauser.Where(toPause => !toPause), 
           _ => pauser.Where(toPause => toPause)) 
         .SelectMany(l => l.ToObservable()); 

    return source.Window(pauser.Where(toPause => toPause).StartWith(true), 
         _ => pauser.Where(toPause => !toPause)) 
       .Switch() 
       .Merge(queue); 
} 

okno jest otwarte na subskrypcję i za każdym razem „prawdziwy” jest odbierane ze strumienia Pauser. Zamyka się, gdy pauser dostarcza "fałszywą" wartość.

Bufor robi to, co powinien, buforuje wartości pomiędzy "false" i "true" z pauser. Gdy Buffer otrzyma "true", wyprowadza IList wartości, które są natychmiastowo przesyłane strumieniowo na raz.

DotNetFiddle Link: https://dotnetfiddle.net/vGU5dJ