在订阅回调中结束订阅

Question

码：

            subscription = source
                // close websocket eventually
                .Finally(() => webSocket.CloseAsync(WebSocketCloseStatus.Empty, String.Empty, CancellationToken.None).Wait())
                .Subscribe(
                    data =>
                    {
                        if (webSocket.State != WebSocketState.Open)
                        {
                            _logger.LogWarning("Websocket closed by client!");
                            // TODO: End subscription
                        }

                        try
                        {
                            webSocket.SendAsync(data.ToString(),
                                WebSocketMessageType.Text, true, CancellationToken.None).Wait();
                        }
                        catch (WebSocketException e)
                        {
                            _logger.LogWarning(e, "problem with websocket!");
                            // TODO: End subscription
                        }
                    });

说明：

• 聆听“来源”中的事件
• 收到数据 - >发送websocket
• 在异常或websocket closed =>应该结束对“source”的订阅并确保websocket已关闭（如果尚未关闭）。怎么样？？

实现这一目标的一种方法是让异常不被处理，这将使观察者停止观察并“最终”调用，但问题是它然后崩溃我的整个服务器，因为异常在某个后台线程上重新抛出并且未处理。

Answer 1

要在回调中结束订阅，您只需在订阅上调用Dispose()：

 IDisposable subscription = null;

 ...


 subscription = source
                // close websocket eventually
                .Finally(() => webSocket.CloseAsync(WebSocketCloseStatus.Empty, String.Empty, CancellationToken.None).Wait())
                .Subscribe(
                    data =>
                    {
                        if (webSocket.State != WebSocketState.Open)
                        {
                            _logger.LogWarning("Websocket closed by client!");
                            subscription.Dispose(); // End subscription
                        }

                        try
                        {
                            webSocket.SendAsync(data.ToString(),
                                WebSocketMessageType.Text, true, CancellationToken.None).Wait();
                        }
                        catch (WebSocketException e)
                        {
                            _logger.LogWarning(e, "problem with websocket!");
                            subscription.Dispose(); // End subscription
                        }
                    });

另见http://www.introtorx.com/content/v1.0.10621.0/03_LifetimeManagement.html#Unsubscribing

Answer 2

你真的应该尝试像这样定义你的observable：

var subscription =
    source
        .TakeWhile(data => webSocket.State == WebSocketState.Open)
        .SelectMany(data =>
            Observable.FromAsync(() =>
                webSocket.SendAsync(data.ToString(), WebSocketMessageType.Text, true, CancellationToken.None)))
        .Catch<WebSocketException>(ex =>
            Observable.FromAsync(() =>
                webSocket.CloseAsync(WebSocketCloseStatus.Empty, String.Empty, CancellationToken.None)))
        .Subscribe();

它最终会自动取消订阅。

即使你可以，你也应该避免在观察者中取消订阅。