Уведомление асинхронного субъекта от другого наблюдаемого с помощью Rx .NET

Question

Я пытаюсь оповестить слушателей, которые подписались на Subject _sub, из другой наблюдаемой, а после этого войти в какое-либо сообщение в обработчике Do.Я звоню OnNext и все будет работать нормально, если _sub не будет асинхронным.Проблема здесь в том, что нет функции OnNextAsync, которую я бы ждал в первой наблюдаемой.Каков наилучший способ сделать это?

 class Program
        {
            private static Subject<int> _sub = new Subject<int>();

            static void Main(string[] args)
            {
                _sub.SelectMany(async _ =>
                {
                    Console.WriteLine("SUB START: " + _);
                    await Task.Delay(3000);
                    Console.WriteLine("SUB END: " + _);
                    return 1;
                }).Subscribe();


                Start();
            }

            public static void Start()
            {
                int count = 0;
                Observable.Interval(TimeSpan.FromSeconds(5)).Select(x =>
                {
                    Console.WriteLine("START INTERVAL");
                    _sub.OnNext(count++); //onNext is not awaitable
                    Console.WriteLine("END INTERVAL");
                    return 1;
                })
                .Do(_ => Console.WriteLine("ALL FINISHED"))
                .Subscribe();

                Console.WriteLine("READLINE");
                Console.ReadLine();
            }

        }

Результат:

READLINE
START INTERVAL
SUB START: 0
END INTERVAL
ALL FINISHED
SUB END: 0

Ожидаемый результат:

READLINE
START INTERVAL
SUB START: 0
SUB END: 0
END INTERVAL
ALL FINISHED

Answer 1

Я собираюсь повторить Пауло для ясности:

• Наблюдаемые не должны заботиться о наблюдателях .В то время как Observable будет ожидать своих синхронных наблюдателей, лучше всего рассматривать это как случайность реализации.Наблюдаемое вообще не ждет от асинхронного наблюдателя.В любом случае, на ожидание не следует полагаться.

• Вы должны действительно переосмыслить, как вы это делаете. Вы используете реактивную библиотекунаписать интерактивный код.Возможно, неправильный выбор инструмента или неправильное использование инструмента.

• Ваш код замусорен Rx code-smells.Думайте о реактивном коде как о блок-схеме.Блок-схема вашего кода будет выглядеть как спагетти.Это должно выглядеть больше как двоичное дерево.

Это звучит как XY проблема : я бы предложил перефразировать ваш вопрос с тем, что вы пытаетесь достичь.

Answer 2

Наблюдаемое не должно полагаться на поведение своих наблюдателей.

Я предлагаю вам переосмыслить все это.То, что вы делаете, выглядит более интерактивным, чем реактивным.