Использование одной наблюдаемой для синхронизации другой наблюдаемой - PullRequest
Новая
       30

Использование одной наблюдаемой для синхронизации другой наблюдаемой

2 голосов
/ 25 апреля 2019

Я хочу синхронизировать наблюдаемое, используя другое, используемое в качестве часов, ниже приведен пример. 

Main:     ---------abc----------------------------------
Clock:    -x-----x-----x-----x-----x-----x-----x-----x--
Expected: -------------a-----b-----c--------------------

Я попытался добиться этой синхронизации, используя метод Zip, аналогично примеру, описанному в документации RX (http://reactivex.io/documentation/operators/zip.html):

mainValues.Zip(clockValues, (mainValue,clockValue) => mainValue)

проблема в том, что когда я тестировал эту реализацию, она не работала. Ниже приведен тест, который я написал для проверки ожидаемого поведения: 

scheduler = new TestScheduler();


var mainValues = scheduler.CreateHotObservable(
    new Recorded<Notification<char>>(100, Notification.CreateOnNext('a')),
    new Recorded<Notification<char>>(101, Notification.CreateOnNext('b')),
    new Recorded<Notification<char>>(102, Notification.CreateOnNext('c')),
    new Recorded<Notification<char>>(103, Notification.CreateOnNext('d')),
    new Recorded<Notification<char>>(104, Notification.CreateOnNext('e')),
    new Recorded<Notification<char>>(105, Notification.CreateOnNext('f')),
    new Recorded<Notification<char>>(106, Notification.CreateOnNext('g')),
    new Recorded<Notification<char>>(107, Notification.CreateOnCompleted<char>()));

var clockValues = scheduler.CreateHotObservable(
    new Recorded<Notification<long>>(70, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(90, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(110, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(130, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(150, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(170, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(190, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(210, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(230, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(250, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(270, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(290, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(310, Notification.CreateOnCompleted<long>()));


var res = scheduler.Start(() => mainValues.Zip(clockValues, (mainValue, clockValue) => mainValue), 0, 70, long.MaxValue);

А вот ниже ожидаемых значений и того, что я действительно получил (описано как комментарий): 

res.Messages.AssertEqual(
    OnNext(110, 'a'), // Expected: 110, a - Actual: 100, a
    OnNext(130, 'b'), // Expected: 130, b - Actual: 110, b
    OnNext(150, 'c'), // Expected: 150, c - Actual: 130, c
    OnNext(170, 'd'), // Expected: 170, d - Actual: 150, d
    OnNext(190, 'e'), // Expected: 190, e - Actual: 170, e
    OnNext(210, 'f'), // Expected: 210, f - Actual: 190, f
    OnNext(230, 'g'));// Expected: 230, g - Actual: 210, g

В чем проблема? Правильно ли использовать Zip для синхронизации двух наблюдаемых? Я неправильно использую TestScheduler?

Ответы [ 2 ]

2 голосов
/ 29 апреля 2019

Дайте ему попробовать: 

var query =
    Observable
        .Create<char?>(o =>
        {
            IDisposable inner = null;
            IDisposable subscription = 
                mainValues
                    .Publish(mvs =>
                    {
                        var q = new System.Collections.Generic.Queue<char>();
                        inner = mvs.Subscribe(mv => q.Enqueue(mv));
                        return clockValues.Select(x => q.Count > 0 ? q.Dequeue() : (char?)null);
                    })
                    .Subscribe(o);
            return new CompositeDisposable(inner, subscription);
        });

query.Subscribe(x => Console.WriteLine(x));
scheduler.Start();

Дайте мне знать, если это работает так, как вы этого хотите.Если это произойдет, я приведу некоторые объяснения.

1 голос
/ 25 апреля 2019

Меняется ли 

mainValues.Zip(clockValues, (mainValue,clockValue) => mainValues)

... чтобы: 

mainValues.Zip(clockValues, (mainValue,clockValue) => mainValue)

это исправить?

...