c# - .NET 的 RX - 从 Disposable.Create 中调用 observer.OnCompleted 时没有任何反应

Question

我正在使用 C# 中的 .NET RX 库。谁能向我解释为什么“observer.OnCompleted()”方法在以下代码中什么都不做:

var observableStream = Observable.Create<CustomMessage>(
       (observer) =>
           {
               CustomMessage cm = new CustomMessage();
               CustomMessage.Subscribe(observer.OnNext);

               return Disposable.Create(
               () =>
                   {
                       Console.WriteLine("Disposing...");
                       CustomMessage.Unsubscribe(observer.OnNext);                          
                       observer.OnCompleted();                //***Nothing happens here***
                   }
               );
       });

    //IObserver.OnException()
    public override void OnException(Exception e)
    {
        Console.WriteLine("Exception occurred - " + e.Message);
    }

    //IObserver.OnComplete()
    public override void OnUnsubscribe()
    {
        Console.WriteLine("Unsubscribed...");            
    }

    //IObserver.OnNext()
    public override void HandleNextMsg(IRVMessage msg)
    {
        Console.WriteLine("Instance received a message");
    }

IDisposable myDisposable = observableStream.Subscribe(HandleNextMsg, OnException, OnUnsubscribe);

//At some later point....
myDisposable.Dispose();

该代码旨在订阅 CustomMessages 流。它在设置订阅时使用我的 CustomMessage 类型注册了 observer.OnNext() 方法。然后在处理订阅时注销 observer.OnNext()。所有这些都正常工作。每当收到 CustomMessage 时，我的“HandleNextMsg()”方法就会被调用。

稍后，当我希望终止我的订阅时，我调用“Dispose()”并成功执行了以下两行:

Console.WriteLine("Disposing...");
CustomMessage.Unsubscribe(observer.OnNext); 

然后我不再收到 CustomMessages。然而，以下行虽然已执行，但什么也不做:

observer.OnCompleted(); 

我希望它调用这条线:

Console.WriteLine("Unsubscribed...");

在某些时候，观察者与“OnUnsubscribe”方法之间的连接丢失了，我想确切地了解发生了什么。 “observer.OnNext()”可以成功注销，但“observer.OnCompleted()”什么都不做，这是怎么回事？

有人向我指出，仅仅因为我正在处理流并不意味着我应该调用“OnCompleted()”，但我仍然想了解为什么它不起作用。

Answer 1

OnCompleted() 旨在通知订阅者上游序列(在您的情况下为 CustomMessage)已结束。这并不意味着确认订阅者请求取消订阅已成功，这似乎是您尝试使用它的方式。 OnCompleted() 是关于序列的通知，针对所有订阅者，而不是针对该序列的单个订阅者。

换句话说，您不应该在 Dispose 中调用它。毕竟，订阅者自己正在取消订阅，为什么需要通知它？

至于什么都没有发生的实际技术原因，我猜测回调(按设计)不会在您已经处理时执行。只是一个理论，它不是很相关。