Здравствуйте, TK, Вы писали:

TK>1. Порядок даст Concat() но, пока идет ожидание будет копиться очередь из не обработанных событий (ну и с TimeOut будут заморочки). Merge() такую очередь копить не будет но, порядок следования событий будет "сбит".
Тут то что надо.

Хочется ещё сделать небольшую оптимизацию, чтобы не ждать большой промежуток а проверять короткими промежутками и выйти по истечению времени если ничего не получилось.

Скажем такой пример:

using System;
using System.Reactive.Linq;

namespace RxTest
{
    class Program
    {
        static void Main(string[] args)
        {
            var r = Observable.Range(0, 10);

            r.Select(x => x)
                .Select(x => getY(x))
                .Concat()
                .Subscribe(x => Console.WriteLine("a: " + x));
            

            Console.ReadKey();
        }

        private static IObservable<int> getY(int i)
        {
            if (i > 5)
            {
                int y;
                if (tryGetReal(i, out y))
                {
                    // No delay
                    return Observable.Return(y);
                }
                else
                {
                    // Delay maximum 300ms with test every 100ms
                    // Как тут сделать ?
                    for (var k = 0; k < 3; k++)
                    {
                        System.Threading.Thread.Sleep(100);
                        if (tryGetReal(i, out y)) return Observable.Return(y);
                    }

                    return Observable.Empty<int>();
                }
            }
            else
            {
                return Observable.Empty<int>();
            }
        }

        private static bool tryGetReal(int i, out int y)
        {
            // Demo code
            y = i;
            return false;
        }
    }

Я так понимаю надо как-то из Defer вызывать Delay рекурсивно но с ограничением количества вызовов но не могу понять как.

Здравствуйте, TK, Вы писали:

TK>1. Порядок даст Concat() но, пока идет ожидание будет копиться очередь из не обработанных событий (ну и с TimeOut будут заморочки). Merge() такую очередь копить не будет но, порядок следования событий будет "сбит".
Тут то что надо.

Хочется ещё сделать небольшую оптимизацию, чтобы не ждать большой промежуток а проверять короткими промежутками и выйти по истечению времени если ничего не получилось.

Скажем такой пример:

using System;
using System.Reactive.Linq;

namespace RxTest
{
    class Program
    {
        static void Main(string[] args)
        {
            var r = Observable.Range(0, 10);

            r.Select(x => x)
                .Select(x => getY(x))
                .Concat()
                .Subscribe(x => Console.WriteLine("a: " + x));
            

            Console.ReadKey();
        }

        private static IObservable<int> getY(int i, int tries = 0)
        {
            Console.WriteLine("In getY: i = {0}, tries = {1}", i, tries);

            if (i > 5)
            {
                int y;
                if (tryGetReal(i, out y))
                {
                    // No delay
                    return Observable.Return(y);
                }
                else
                {
                    if (tries < 3)
                    {
                        // Let's Try again
                        return Observable.Defer(() => getY(i, tries + 1)).Delay(TimeSpan.FromMilliseconds(100));
                    }
                    else
                    {
                        // No more tries, stop !
                        return Observable.Empty<int>();
                    }
                }
            }
            else
            {
                return Observable.Empty<int>();
            }
        }

        private static bool tryGetReal(int i, out int y)
        {
            if (i > 7)
            {
                y = i;
                return false;
            }
            else
            {
                y = i;
                return true;
            }
        }
    }
}

Насколько это правильно ?