逐个事件完成

Question

使用Observable.FromEvent可观察到的事件有一种方法。例如。这个课程：

class Generator<T>
{
    event Action<T> onPush;
    public IObservable<T> Items =>
        Observable.FromEvent<T>(d => onPush += d, d => onPush -= d);
    public void Push(T item) => onPush?.Invoke(item);
}

但是，我还没有找到一种通过事件完成可观察到的方法 - 我该怎么做？

更新：

为了澄清我的意思，上面的类产生的IObservable<T>是"无尽"而从未完成的。我想通过另一个事件完成它，而不是让另一个可观察到。因此，问题可以减少：

如何提前完成任意IObservable<T>，即OnCompleted通知？

Answer 1

可观察到的通知或事件流。当一个事件的可观察到的来源时，它们本质上是无尽的。可观察的可连接到事件，引用对象，因此备份事件的对象永远不会超出范围。.NET/C＃没有提供一种表明事件永远不会再次调用的方法，因此可观察到的直接连接到该事件的可观察到无限。

这并不少见；大多数基于事件的可观察结果永远不会显式地称为OnCompleted，对现实世界进行建模，很难确定地说某些事情将永远不会再次发生。

但是，这不是一个问题：可观察到的意图是无限运行，并且不会造成损害。未订阅的可观察到的并不占用太多资源。如果您对可观察到的活动不感兴趣，请退订所有订阅，并且很好。

一种方法是与Take运算符之一，例如TakeUntil操作员（如下所述）。尝试以下代码（使用您的Generator类）：

var g = new Generator<int>();
g.Items
    .TakeUntil(i => i > 3)
    .Subscribe(
        i => Console.WriteLine($"OnNext: {i}"), 
        e => Console.WriteLine($"OnError: Message: {e.Message}"), 
        () => Console.WriteLine("OnCompleted")
    );
g.Push(1);
g.Push(2);
g.Push(3);
g.Push(4);
g.Push(5);
g.Push(6);

输出：

OnNext: 1
OnNext: 2
OnNext: 3
OnNext: 4
OnCompleted

TakeUntil在有大于3的整数的消息后观察到的Items可观察到。这就是为什么有一个on completed andcompleted和5，6个消息的原因。

另外，正如提到的那样，您的Generator<T>类基本与Subject<T>相同，我建议您使用它。

---

原始答案：

使事件中的另一个可观察到，然后使用.TakeUntil：

class Generator<T>
{
    event Action<T> onPush;
    event Action<Unit> onCompleted;
    public IObservable<T> Items =>
        Observable.FromEvent<T>(d => onPush += d, d => onPush -= d)
        .TakeUntil(Completion);
    public IObservable<Unit> Completion =>
        Observable.FromEvent<Unit>(d => onCompleted += d, d => onCompleted -= d);
    public void Push(T item) => onPush?.Invoke(item);
    public void Complete() => onCompleted?.Invoke(Unit.Default);
}