Jak naprzemiennie buforować i przepuszczać strumień danych na żywo w Rx

Question

Mam dwa strumienie. Jednym z nich jest przepływ danych (może to być dowolny typ), drugi to strumień boolowski działający jako bramka. Muszę połączyć je w strumieniu, który ma następujące zachowanie:

• Kiedy brama jest otwarta (najnowsza wartość była prawda), wówczas dane powinien płynąć prosto przez
• Kiedy brama jest zamknięta (większość ostatnie wartości są nieprawdziwe), a następnie dane powinny być buforowane być wydany, jak poszczególne elementy, gdy brama znajduje się obok otwarty
• roztwór powinien zachować wszystkich elementów danych, przy czym zachowanie celu

Nie jestem pewien, jak to połączyć. Wejścia I zostały badania ze to tak:

// a demo data stream that emits every second 
var dataStream = Observable.Interval(TimeSpan.FromSeconds(1)); 

// a demo flag stream that toggles every 5 seconds 
var toggle = false; 
var gateStream = Observable.Interval(TimeSpan.FromSeconds(5)) 
    .Select(_ => toggle = !toggle); 

Answer 1

Chciałbym to zrobić w następujący sposób:

• Okno strumień danych za pomocą strumienia brama jako selektor zamykania
• Możemy użyć DistinctUntilChanged na strumieniu bramki, aby zapewnić brak powtarzających się wartości
• Zmusimy także, aby strumień bramki zaczął się zamknięty (fałsz) - nie wpłynie to na wyjście i pozwala na schludną sztuczkę
• T użyj przeciążenia Select, które nadaje każdemu elementowi numer indeksu. Dzięki temu możemy stwierdzić, czy musimy buforować, czy po prostu emitować okno, ponieważ wiemy, że okna o numerach parzystych służą do buforowania (ponieważ upewniliśmy się, że strumień bramki zaczyna się od fałszu).
• Możemy użyć ToList() do buforowania każdy nawet okno, aż zamknie się - to jest rzeczywiście odpowiednikiem Buffer() że czeka aż OnCompleted
• używamy tożsamość SelectMany spłaszczyć buforowane okna
• Wreszcie możemy złączyć okna, aby zapewnić porządek jest zachowany

Wygląda to tak:

dataStream.Window(gateStream.StartWith(false).DistinctUntilChanged()) 
      .Select((w, i) => i % 2 == 0 ? w.ToList().SelectMany(x => x) : w) 
      .Concat() 
      .Subscribe(Console.WriteLine);