Czy mogę sparować dwie sekwencje za pomocą pasującego klucza?

Question

Powiedzmy, że pierwsza sekwencja wychodzi do sieci, aby pobrać zawartość stron 1, 2, 3, 4, 5 (ale powróci w nieprzewidywalnej kolejności).

Sekwencja druga polega na przejściu do bazy danych w celu pobrania kontekstu dotyczącego tych samych rekordów 1, 2, 3, 4, 5 (ale dla celów tego przykładu nastąpi powrót w nieprzewidywalnej kolejności).

Czy istnieje metoda rozszerzenia Rx, która połączy je w jedną sekwencję , gdy każda pasująca para jest gotowa w obu sekwencjach? Oznacza to, że jeśli pierwsza sekwencja powróci w kolejności 4,2,3,5,1, a druga sekwencja powróci w kolejności 1,4,3,2,5, połączona sekwencja będzie wynosić (4,4), (3 , 3), (2,2), (1,1), (5,5) - zaraz po przygotowaniu każdej pary. Spojrzałem na Merge i Zip, ale nie wydają mi się dokładnie tym, czego szukam.

Nie chciałbym odrzucać par, które nie pasują, co moim zdaniem wyklucza prostą kombinację .Where.Select.

Answer 1

var paired = Observable 
    .Merge(aSource, bSource) 
    .GroupBy(i => i) 
    .SelectMany(g => g.Buffer(2).Take(1)); 

Poniższy test daje prawidłowe wyniki. W tej chwili używamy właśnie ints, jeśli używasz danych z kluczami i wartościami, musisz pogrupować według i.Key zamiast i.

var aSource = new Subject<int>(); 
var bSource = new Subject<int>(); 

paired.Subscribe(g => Console.WriteLine("{0}:{1}", g.ElementAt(0), g.ElementAt(1))); 

aSource.OnNext(4); 
bSource.OnNext(1); 
aSource.OnNext(2); 
bSource.OnNext(4); 
aSource.OnNext(3); 
bSource.OnNext(3); 
aSource.OnNext(5); 
bSource.OnNext(2); 
aSource.OnNext(1); 
bSource.OnNext(5); 

plony:

4:4 
3:3 
2:2 
1:1 
5:5 

Edycja w odpowiedzi Brandon:

Dla sytuacji, gdy produkty te są różne klasy (AClass i BClass), następujące korekty mogą być wykonane.

using Pair = Tuple<AClass, BClass>; 

var paired = Observable 
    .Merge(aSource.Select(a => new Pair(a, null)), bSource.Select(b => new Pair(null, b))) 
    .GroupBy(p => p.Item1 != null ? p.Item1.Key : p.Item2.Key) 
    .SelectMany(g => g.Buffer(2).Take(1)) 
    .Select(g => new Pair(
     g.ElementAt(0).Item1 ?? g.ElementAt(1).Item1, 
     g.ElementAt(0).Item2 ?? g.ElementAt(1).Item2)); 

Answer 2

Więc masz 2 obserwowalne sekwencje, które chcesz sparować razem?

Para z numeru Rxx wraz z GroupBy może pomóc tutaj. Myślę kod podobny do poniższego może robić co chcesz

var pairs = stream1.Pair(stream2) 
     .GroupBy(pair => pair.Switch(source1 => source1.Key, source2 => source2.Key)) 
     .SelectMany(group => group.Take(2).ToArray()) // each group will have at most 2 results (1 left and 1 right) 
     .Select(pair => 
     { 
      T1 result1 = default(T1); 
      T2 result2 = default(T2); 

      foreach (var r in pair) 
      { 
       if (r.IsLeft) result1 = r.Left; 
       else result2 = r.Right; 
      } 

      return new { result1, result2 }; 
     }); 

`` `

Nie Przetestowałem go i nie dodaje się w niczym do obsługi błędów, ale myślę, że to, co chcesz .