Add support for unsubscribing from an IObservable<T> streaming result

[Tragetaschen]

This is likely related to #481

I have ported my PersistentConnection from SignalR 2.2 to a Hub method returning an IObservable<T>. By desing, this observable doesn't complete throughout the lifetime of the application.

Whenever a client disconnects (I call connection.stop() in TS), the subscribed observer doesn't unsubscribe and all of them just pile up in my observable. SignalR just loses the subscription:
https://github.com/aspnet/SignalR/blob/rel/1.0.0-alpha1/src/Microsoft.AspNetCore.SignalR.Core/Internal/AsyncEnumeratorAdapters.cs#L39
I've quickly hacked an unsubscribe tied to the Hub's Dispose method:

public class MyHub : Hub
{
    private readonly MyObservable myObservable;

    private IDisposable disposeSubscription;

    public MyHub(MyObservable myObservable) => this.myObservable = myObservable;

    public IObservable<object> Data() => new wrappingObservable(myObservable, this);

    protected override void Dispose(bool disposing)
    {
        disposeSubscription?.Dispose();
        base.Dispose(disposing);
    }

    private class wrappingObservable : IObservable<object>
    {
        private readonly IObservable<object> original;
        private readonly MyHub myHub;

        public wrappingObservable(
            IObservable<object> original,
            MyHub myHub
        )
        {
            this.original = original;
            this.myHub = myHub;
        }

        public IDisposable Subscribe(IObserver<object> observer)
        {
            var disposable = original.Subscribe(observer);
            myHub.disposeSubscription = disposable;
            return disposable;
        }
    }
}

[moozzyk]

This is related to #481 indeed.

[davidfowl]

We could actually have handled this without designing cancellation in the protocol. This is just something we missed. You can tie it to the connection lifetime instead of the hub lifetime.

[davidfowl]

#481 is about cancellation separate from the connection lifetime. Right now we just leak unnecessarily.

[moozzyk]

I think we should have a common mechanism of cancelling streaming which could be invoked in different ways

• on the server - automatically when the connection is stopped
• on the server - on demand by the user
• from the client - by sending a cancellation (which can actually be invoking a hub method that does no. 2 above and we don't need any special cancellation protocol)

[davidfowl]

I don't agree that it should be common. It needs to map to whatever primitive makes sense for the data type.

on the server - automatically when the connection is stopped

I'm super sad we missed this one, maybe we ship an update to the alpha with this fixed. This makes streaming unusable since it's just going to leak memory like crazy. This doesn't require any interaction on behalf of the user.

on the server - on demand by the user

We should just expose a cancellation token that represents the connection lifetime. We used to expose the Transport directly on the connection context (which is too powerful) that allowed you to write code to detect this situation.

from the client - by sending a cancellation (which can actually be invoking a hub method that does no. 2 above and we don't need any special cancellation protocol)

This requires protocol changes to be nice. This shouldn't require an action on the user. I'd rather not hack it with an explicit invocation.

[Tragetaschen]

It goes without saying, but still: Yay, works!