How to correctly dispose a client stream if the connection has disconnected?

Question

I'm using Microsoft.AspNetCore.SignalR 2.1 v1.0.4 and have a ChannelReader stream being consumed by a typescript client using v1.0.4.

The channel surfaces event data specific to a single entity, so it's expected the client would subscribe to a channel when the user of the client navigates to a page rendering data for that single entity. If the user navigates to the same page but for a different entity then the client would make another subscribe call.

Now the questions I have are about how best to unsubscribe the stream, and also in general, 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)?

There doesn't appear to be some connection state surfaced from the api so I currently use an RxJs Subject to surface some connection state to my UI components/services, i.e. when the hub connection's start call is successful I surface "true", and when the onclose callback is called I surface "false". This allows me to attempt to call dispose on a previously subscribed stream to clean things up during a connection disconnect/stop, and then if necessary call subscribe to the stream again on a successful start call.

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. Should I be able to dispose a stream even when the hub is disconnected?

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

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).

When the connection is terminated (either because of stop being called on the client or the server aborting the connection) the error method of your subscriber will be called with an error indicating the stream has been terminated because the connection was terminated. In general, you should handle the error method and consider it a terminal event (i.e. the stream will never yield additional objects). On the server, the Context.ConnectionAborted token will be triggered if the connection is terminated (by either side) and you can stop writing to your stream.

If you're already using RxJS, I'd highly recommend building a small wrapper to convert the object you get back from SignalR into a proper RxJS Observable. The object we return is not actually an Observable, but it has all the same basic methods (a subscribe method that takes an object with complete, next and error methods), so it should be trivial to wrap it.

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.

Yeah, that's probably a bug. We shouldn't throw if you dispose after a hub is disconnected. Can you file that on https://github.com/aspnet/SignalR ? To work around it you can fairly safely just try...catch the error and suppress it (or maybe log it if you're paranoid).

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

You should always dispose the subscription. If you just delete it, then we have no way to know that you're done with it and we never tell the server to stop. If you call dispose (and are connected) we send a message to the server "cancelling" the stream. In ASP.NET Core 2.1 we don't expose this cancellation to you, but we do stop reading from the ChannelReader. In ASP.NET Core 2.2 we allow you to accept a CancellationToken in your Hub method and the dispose method on the client will trigger this token in the Hub method. I'd highly recommend you try the latest preview of ASP.NET Core 2.2 and use a CancellationToken in your Hub method to stop the stream:

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
}

Note: If you do this, you don't need to monitor Context.ConnectionAborted, the token passed in to your Hub method will cover all cancellation cases.

On a related note, you should always use Channel.CreateBounded<T>(size) to create your channel. If you use an unbounded channel it's much easier to leak memory since the writer can just keep writing indefinitely. If you use a bounded channel, the writer will be stopped (WriteAsync and WaitToWriteAsync will "block") if there are size un-read items in the channel (because, for example, the client has disconnected and we've stopped reading).