Вид withLatestFrom обе переменные источники скорости

Question

я следующий случай с результатом я хочу добиться:

L --1---3---4---5---6---7---8- 
R ---A--------B--CDE-FG--H---I 

O ---A--------B--C--E---G---H- 
    ---1--------4--5--6---7---H- 

Так в основном у меня есть два источника с изменением относительной скорости с течением времени. Результат должен игнорировать Иногда L элементов, а иногда R элементов. Как построить такой RX?

Я пытался с помощью Joins, Sample, WithLatestFrom, Distincts и т. Д. Но я всегда могу игнорировать некоторые элементы из L источника, но не R. Я пытаюсь достичь этого в C#, но любой ответ на этот вопрос будет полезен.

Answer 1

В вашем запросе не существует правила. Вы должны иметь возможность последовательно описывать, что должно происходить со значениями от L и R.

Это ближе всего, что я мог бы получить:

var L = new Subject<int>(); 
var R = new Subject<char>(); 

var O = L.Select(l => R.Select(r => new { r, l })).Switch(); 

O.Subscribe(o => Console.WriteLine($"{o.l}{o.r}")); 

L.OnNext(1); 
R.OnNext('A'); 
L.OnNext(3); 
L.OnNext(4); 
R.OnNext('B'); 
L.OnNext(5); 
R.OnNext('C'); 
R.OnNext('D'); 
R.OnNext('E'); 
L.OnNext(6); 
R.OnNext('F'); 
R.OnNext('G'); 
L.OnNext(7); 
R.OnNext('H'); 
L.OnNext(8); 
R.OnNext('I'); 

Это дало мне:

 
1A 
4B 
5C 
5D 
5E 
6F 
6G 
7H 
8I 

Кроме того, из вашей мраморной диаграмме это не ясно, если 'F' или 6 приходит первым. Вам необходимо уточнить порядок значений и правила для объединения потоков.

Answer 2

Кажется, что вы ищете реализацию ZipLatest. См Does my "zipLatest" operator already exist?

EDIT:

Если вы хотите реализацию, без загрузки библиотеки, я думаю, что это будет работать:

public static IObservable<Tuple<T1, T2>> ZipLatest<T1, T2>(this IObservable<T1> left, IObservable<T2> right) 
{ 
    return Observable.Defer(() => 
    { 
     var leftSubject = new BehaviorSubject<bool>(false); 
     var rightSubject = new BehaviorSubject<bool>(false); 

     return left.Publish(_left => 
      right.Publish(_right => 
      { 
       return Observable.CombineLatest(
        _left, 
        _right, 
        _left.Select(_ => true).Merge(leftSubject), 
        _right.Select(_ => true).Merge(rightSubject), 
        (l, r, l_bool, r_bool) => Tuple.Create(l_bool, r_bool, l, r) 
       ) 
        .Where(t => t.Item1 && t.Item2) 
        .Select(t => Tuple.Create(t.Item3, t.Item4)) 
        .Do(_ => 
        { 
         leftSubject.OnNext(false); 
         rightSubject.OnNext(false); 
        }); 
      }) 
     ); 
    }); 

Запуск его от @ тестового кода Enigmativity в (модифицированный) .. .

void Main() 
{ 
    var L = new Subject<int>(); 
    var R = new Subject<char>(); 

    var O = L.ZipLatest(R) 
     .Select(t => new { l = t.Item1, r = t.Item2}); 

    O.Subscribe(o => Console.WriteLine($"{o.l}{o.r}")); 

    L.OnNext(1); 
    R.OnNext('A'); 
    L.OnNext(3); 
    L.OnNext(4); 
    R.OnNext('B'); 
    L.OnNext(5); 
    R.OnNext('C'); 
    R.OnNext('D'); 
    R.OnNext('E'); 
    L.OnNext(6); 
    R.OnNext('F'); 
    R.OnNext('G'); 
    L.OnNext(7); 
    R.OnNext('H'); 
    L.OnNext(8); 
    R.OnNext('I'); 
} 

... вы получите правильные результаты:

1A 
4B 
5C 
6E 
7G 
8H 

Answer 3

Это моя попытка; Я попробую это в компиляторе позже!

Отслеживайте последнее значение, отправленное последним комбайном, и убедитесь, что каждое значение уникально перед применением операции. Определите эту операцию как метод расширения.

public static IObservable<C> ZipLatest<A, B, C>(
    this IObservable<A> sourceA, 
    IObservable<B> sourceB, 
    Func<A, B, C> op) 
{ 
    IObservable<Tuple<A, B>> combined = sourceA.CombineLatest(
      sourceB, (a, b) => Tuple.Create(a, b)); 
    Tuple<A, B> last = null; 
    return combined.Where(curr => 
    { 
     if (last == null || (last.Item1 != curr.Item1 && last.Item2 != curr.Item2)) 
     { 
      last = curr; 
      return true; 
     } 
     return false; 
    }).Select(curr => op(curr.Item1, curr.Item2)); 
}