c# - 如何在订阅开始之前延迟从 IObservable 获取元素？

Question

我尝试在 Observable 序列开始之前插入 1 秒的延迟，但在调用 Subscribe 后立即发出第一条记录。在下面的示例中，我尝试通过传递 1s TimeSpan 来使用 RX 的 Take 运算符，并期望在调用 Subscribe< 时延迟接收数组中的第一个字符串...

public static void Main(string[] args)
{
    IEnumerable<string> e = new[] { "Hi", "There", "Bye" };
    IObservable<string> strings = e.ToObservable();

    IObservable<string> stringsTimed = strings.Take(TimeSpan.FromMilliseconds(1000));
    stringsTimed.Trace("string");

    Console.ReadLine();
}

public static IDisposable Trace<T>(this IObservable<T> source, string name)
{
    return source.Subscribe
    (
        onNext: t => Console.WriteLine($"{name} -> {t}"),
        onError: ex => Console.WriteLine($"{name} ERROR: {ex.Message}"),
        onCompleted: () => Console.WriteLine($"{name} END")
    );
}

但是当我运行程序时，“Hi”、“There”、“Bye”的结果立即打印在屏幕上，没有任何延迟，那么如何在接收第一个元素“Hi”之前添加 1 秒的延迟？

Answer 1

Take 不会延迟订阅。 Take 的 TimeSpan 重载设置了可观察对象在继续之前观察/获取记录的持续时间。

尝试下面的代码。如果您想使用帖子中的扩展方法(而不是下面的 RX Subscribe 扩展方法) - 在 DelaySubscription...

之后添加分号

string name = "Sample";
IEnumerable<string> e = new[] { "Hi", "There", "Bye" };

e.ToObservable()
    .DelaySubscription(TimeSpan.FromMilliseconds(1000))
    .Subscribe( onNext: t => Console.WriteLine($"{name} -> {t}"),
                onError: ex => Console.WriteLine($"{name} ERROR: {ex.Message}"),
                onCompleted: () => Console.WriteLine($"{name} END"));