按事件完成 IObservable

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);

}

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

更新：

为了澄清我的意思，上面的课程产生IObservable<T>了“无尽的”并且永远不会完成。我想让它由另一个事件完成，而不是另一个可观察的。所以问题可以简化为：

如何使任意一个IObservable<T>过早完成，即被OnCompleted调用的通知？

Answer 1

一个 observable 代表一个通知流或事件流。当一个可观察的来源来自一个事件时，它们本质上是无穷无尽的。observable 连接到事件，引用对象，因此支持事件的对象永远不会超出范围。.NET/C# 没有提供一种方法来指示一个事件将不再被调用，因此直接连接到该事件的 observable 是无穷无尽的。这并不少见；大多数基于事件的可观察对象从未OnCompleted明确调用过，而是对现实世界进行建模，在这个世界中很难明确地说某些事情永远不会再发生。然而，这不是问题：Observables 意味着无限运行，并且不会造成任何损害。未订阅的 observable 不会占用太多资源。如果你对事件源 observable 不感兴趣，取消订阅所有订阅就可以了。一种方法是使用其中一个Take运算符，例如TakeUntil运算符（如下所述）。尝试以下代码（使用您的Generator课程）：var g = new Generator<int>();g.Items&nbsp; &nbsp; .TakeUntil(i => i > 3)&nbsp; &nbsp; .Subscribe(&nbsp; &nbsp; &nbsp; &nbsp; i => Console.WriteLine($"OnNext: {i}"),&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; e => Console.WriteLine($"OnError: Message: {e.Message}"),&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; () => Console.WriteLine("OnCompleted")&nbsp; &nbsp; );g.Push(1);g.Push(2);g.Push(3);g.Push(4);g.Push(5);g.Push(6);输出：OnNext: 1OnNext: 2OnNext: 3OnNext: 4OnCompletedTakeUntil在有整数大于 3 的消息后取消订阅Itemsobservable。这就是为什么有 OnCompleted，没有 5、6 条消息的原因。另外，正如 Enigmativity 所提到的，您的Generator<T>课程与 基本相同Subject<T>，我建议您使用它。原答案：从事件中制作另一个可观察的，然后使用.TakeUntil：class Generator<T>{&nbsp; &nbsp; event Action<T> onPush;&nbsp; &nbsp; event Action<Unit> onCompleted;&nbsp; &nbsp; public IObservable<T> Items =>&nbsp; &nbsp; &nbsp; &nbsp; Observable.FromEvent<T>(d => onPush += d, d => onPush -= d)&nbsp; &nbsp; &nbsp; &nbsp; .TakeUntil(Completion);&nbsp; &nbsp; public IObservable<Unit> Completion =>&nbsp; &nbsp; &nbsp; &nbsp; Observable.FromEvent<Unit>(d => onCompleted += d, d => onCompleted -= d);&nbsp; &nbsp; public void Push(T item) => onPush?.Invoke(item);&nbsp; &nbsp; public void Complete() => onCompleted?.Invoke(Unit.Default);}