asp.net-core-signalr - 如果连接断开，如何正确处理客户端流？

Question

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

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

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

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

我尝试在流上调用 dispose，如果集线器已连接，则没问题，但如果连接处于断开状态，则会出错。我想知道这是否是一个错误。即使集线器断开连接，我是否也应该能够处理流？

是否可以只执行删除流订阅，然后根据需要重新创建，或者这会以任何方式泄漏吗？

Answer 1

what the lifetime of the stream is to the client under hub connection stop/start scenarios, and if the server explicitly aborts a connection (due to access_token timeouts and so to trigger the client to refresh their connection).

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

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

I have tried calling dispose on a stream which is fine if the hub is connected, but it errors if the connection is in a disconnected state. I'm wondering if this is a bug or not.

是的，这可能是一个错误。如果您在集线器断开连接后丢弃，我们不应该扔掉。您可以将其归档到 https://github.com/aspnet/SignalR ？要解决这个问题，您可以相当安全地 try...catch错误并抑制它(或者如果你偏执的话可以记录它)。

Is it okay to just do a delete streamsubscription and then recreate as required, or will this leak in any way?

您应该始终 dispose订阅。如果你只是delete它，那么我们就无法知道您已完成它，并且我们永远不会告诉服务器停止。如果您调用dispose (并且已连接)我们向服务器发送一条消息“取消”流。在 ASP.NET Core 2.1 中，我们不会向您公开此取消，但我们确实停止从 ChannelReader 读取内容。 。在 ASP.NET Core 2.2 中，我们允许您接受 CancellationToken在您的 Hub 方法和 dispose 中客户端上的方法将在 Hub 方法中触发此 token 。我强烈建议您尝试 ASP.NET Core 2.2 的最新预览版并使用 CancellationToken在您的 Hub 方法中停止流:

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 方法的 token 将涵盖所有取消情况。

在相关说明中，您应该始终使用 Channel.CreateBounded<T>(size)创建您的 channel 。如果您使用无界 channel ，则更容易泄漏内存，因为编写器可以无限期地继续写入。如果您使用有界 channel ，则当存在 WriteAsync 时，写入器将被停止( WaitToWriteAsync 和 size 将“阻塞”)。 channel 中未读的项目(例如，因为客户端已断开连接，我们已停止阅读)。