如果连接断开，如何正确处理客户端流？

Question

我正在使用 Microsoft.AspNetCore.SignalR 2.1 v1.0.4，并且使用 v1.0.4 的打字稿客户端正在使用 ChannelReader 流。

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

现在我的问题是如何最好地取消订阅流，并且一般来说，在集线器连接停止/启动场景下流的生命周期对于客户端来说是多少，以及服务器是否显式中止连接（由于 access_token超时等触发客户端刷新连接）？

API 似乎没有显示某些连接状态，因此我当前使用 RxJs 主题来显示我的 UI 组件/服务的某些连接状态，即，当集线器连接的启动调用成功时，我显示“true”，并且当调用 onclose 回调时，我显示“false”。这允许我尝试在连接断开/停止期间对先前订阅的流调用 dispose 来清理内容，然后在必要时在成功启动调用时再次调用 subscribe 来订阅流。

我尝试在流上调用 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 Core 2.1 中，我们不会向您公开此取消，但我们会停止从

ChannelReader

读取内容。在 ASP.NET Core 2.2 中，我们允许您在 Hub 方法中接受

CancellationToken

，并且客户端上的

dispose

方法将在 Hub 方法中触发此令牌。我强烈建议您尝试 ASP.NET Core 2.2 的最新预览版，并在 Hub 方法中使用

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

，传入您的 Hub 方法的令牌将涵盖所有取消情况。

相关说明，您应该

始终

使用Channel.CreateBounded<T>(size)来创建频道。如果您使用无界通道，则更容易泄漏内存，因为编写器可以无限期地继续写入。如果您使用有界通道，如果通道中有

WriteAsync

未读项目（因为，例如，客户端已断开连接，并且我们已经停止阅读）。