如果连接断开，如何正确地配置客户端流？

Question

我使用的是Microsoft.AspNetCore.SignalR 2.1v1.0.4，一个类型记录客户机使用v1.0.4来使用ChannelReader流。

通道会显示特定于单个实体的事件数据，因此当客户端的用户导航到该实体的页面呈现数据时，预期客户端将订阅该通道。如果用户导航到同一页，但指向不同的实体，则客户端将进行另一次订阅调用。

现在，我要问的问题是，如何最好地取消订阅流，以及一般而言，在集线器连接停止/启动方案下，流的生存期是什么?如果服务器显式中止连接(由于access_token超时而触发客户机刷新连接)？

api中似乎没有出现某种连接状态，因此我目前使用一个RxJ来显示到我的UI组件/服务的某些连接状态，即当集线器连接的开始调用成功时，I表面"true"，以及当onclose回调被称为I surface "false“时。这允许我尝试在先前订阅的流上调用dispose，以便在连接断开/停止期间清理事物，如果有必要，还可以在成功的开始调用时再次调用订阅流。

我已经尝试过在流上调用dispose，如果集线器被连接，这是很好的，但是如果连接处于断开状态，它就会出错。我想知道这是不是个窃听器。即使集线器断开，我也应该能够释放流吗？

只做一个delete streamsubscription，然后根据需要重新创建，还是会以任何方式泄漏？

Answer 1

在集线器连接停止/启动场景下，以及如果服务器显式中止连接(由于access_token超时而触发客户端刷新其连接)，流的生存期将如何？

当连接终止时(因为客户端上调用stop或服务器中止连接)，您的订阅服务器的error方法将被调用，其中一个错误指示由于连接被终止而终止了流。通常，您应该处理error方法，并将其视为一个终端事件(即流永远不会产生额外的对象)。在服务器上，如果连接被终止(任何一方)，并且您可以停止写入流，Context.ConnectionAborted令牌将被触发。

如果您已经在使用RxJS，我强烈建议构建一个小包装器，将从SignalR返回的对象转换为适当的RxJS Observable。我们返回的对象实际上不是一个Observable，但是它有所有相同的基本方法(一个subscribe方法，它使用带有complete、next和error方法的对象)，所以包装它应该很简单。

我已经尝试过在流上调用dispose，如果集线器被连接，这是很好的，但是如果连接处于断开状态，它就会出错。我想知道这是不是个窃听器。

是啊那可能是个窃听器。如果你在集线器断开后处理掉，我们不应该抛出。你能把它放到https://github.com/aspnet/SignalR上吗？要解决这个问题，您可以非常安全地try...catch错误并抑制它(如果您偏执的话，也可以记录它)。

只做一个删除流订阅，然后根据需要重新创建，还是会以任何方式泄漏？

您应该始终 dispose订阅。如果您只是delete它，那么我们就无法知道您已经完成了它，我们从来没有告诉服务器停止。如果您调用dispose (并已连接)，我们将向服务器发送一条“取消”流的消息。在ASP.NET Core2.1中，我们不向您公开此取消，但我们确实停止了从ChannelReader中读取。在ASP.NET Core2.2中，我们允许您在集线器方法中接受CancellationToken，而客户机上的dispose方法将在集线器方法中触发此令牌。我强烈建议您尝试使用ASP.NET Core2.2的最新预览，并在集线器方法中使用CancellationToken来停止流：

public ChannelReader<object> MyStreamingMethod(..., CancellationToken cancellationToken) {
    // pass 'cancellationToken' over to whatever process is writing to the channel
    // and stop writing when the token is triggered
}

注意:如果您这样做，您不需要监视Context.ConnectionAborted，传入您的集线器方法的令牌将涵盖所有取消情况。

另外，您应该始终使用使用Channel.CreateBounded<T>(size)来创建您的频道。如果您使用一个无限制的通道，那么泄漏内存就容易多了，因为作者可以无限期地写下去。如果使用有界通道，如果通道中存在WriteAsync bounded项，则将停止写入器( WaitToWriteAsync和size将“阻塞”)(例如，客户端已断开连接，而我们已停止读取)。