Возникли проблемы с горячей наблюдаемой - PullRequest
Новая
       16

Возникли проблемы с горячей наблюдаемой

1 голос
/ 21 июня 2019

Я пытаюсь создать два конвейера для каждого символа. Каждый символ будет Buffer данными, основанными на двух таймфреймах, и будет выполнять DoCalc на каждом таймфрейме. 

priceChangedObservable = Observable.FromEvent<QuoteChangeEvent, IQuote>(handler =>
    {
        QuoteChangeEvent qHandler = (e) =>
        {
            handler(e);
        };

        return qHandler;
    },
    qHandler => bapi.MAPI.OnQuoteChange += qHandler,
    qHandler => bapi.MAPI.OnQuoteChange -= qHandler
    );

Если я сделаю следующее: 

var els = new EventLoopScheduler();
var dispatcher = new EventLoopScheduler();

   var multiCastStream = Observable.Publish(priceChangedObservable);

    int timeFrame = 60;
     multiCastStream
    .GroupBy(instrument => instrument.Symbol)
    .SelectMany(q => q)            
    .Buffer(TimeSpan.FromSeconds(timeFrame))
    .Where(messages => messages.Any())
    .SubscribeOn(els)
    .ObserveOn(dispatcher)
    .Select((sr) => DoCalc(sr, timeFrame))
    .Subscribe((en) => { if (null != en) Console.WriteLine(en); });

    // Start the producer
    multiCastStream.Connect();

Все работает так, как я ожидаю. Если я закомментирую приведенный выше код и добавлю второй таймфрейм перед оператором multiCastStream.Connect(): 

    int secondTimeFrame = 300;
    multiCastStream
    .GroupBy(instrument => instrument.Symbol)
    .SelectMany(q => q)
    .Buffer(TimeSpan.FromSeconds(secondTimeFrame))
    .Where(messages => messages.Any())
    .SubscribeOn(els)
    .ObserveOn(dispatcher)
    .Select((sr) => DoCalc(sr, secondTimeFrame))
    .Subscribe((en) => { if (null != en) Console.WriteLine(en); });

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

Есть ли что-то фундаментальное, что я упускаю, когда делюсь Hot Observables?

РЕДАКТИРОВАТЬ 1

После изменения кода с ответом от Aron я получаю: 

Number of quotes 1
60: 6/20/2019 10:53:26 PM=> M2KU9 Stats.
Number of quotes 1
300: 6/20/2019 10:53:26 PM=> M2KU9 Stats.
Number of quotes 40
60: 6/20/2019 10:54:26 PM=> MNQU9 Stats.

Тогда никакая другая статистика не будет напечатана.

1 Ответ

1 голос
/ 21 июня 2019

Все работает как положено, ваша логика ошибочна.Для демонстрации я нарисую диаграмму Rx Marble 

Quote   -----------x--------------x-----------x-----------x--
Buffer1 |   |   |   |   |   |   |   |   |   |   |   |   |   |
Buffer2   |       |       |       |       |       |       |   
Where 1 ------------x---------------x-----------x-----------x
Where 2 ----------------------------------x-------x-----------
                    ^Notice It seems like there should be an event on 2
                     But there wasn't?

Вместо этого вы могли бы синхронизировать эти 

var synchronizationSource = Observerable.Timer(TimeSpan.FromSeconds(1),TimeSpan.FromSeconds(1))
     .Publish()
using(synchronizationSource.Connect())
{
       IObservable<Stat> CreateTimeFrame(int seconds)
       {
           var bufferTimeFrame = synchronizationSource
                         .Where(i => i % seconds == 0);
           return priceChangedObservable
               .GroupBy(instrument => instrument.Symbol)
               .SelectMany(q => q)            
               .Buffer(bufferTimeFrame)
               .Where(messages => messages.Any())
               .ObserveOn(dispatcher)
               .Select((sr) => DoCalc(sr, timeFrame))
       }

       CreateTimeFrame(60)
          .Subscribe((en) => { if (null != en) Console.WriteLine(en); });
       CreateTimeFrame(300)
          .Subscribe((en) => { if (null != en) Console.WriteLine(en); });


       Thread.Sleep(100000);
 }
...