Get list of active subscriptions

[KjellOlav]

We use GraphQL subscriptions to show live data, and we use a cloud service to feed live data to these subscriptions. Since cloud services are paid for based on usage, we would like to feed the graphql subscriptions only when there are someone watching. So if a user opens one of the web sites that contains subscriptions, this would activate the "data feeder".

To do that I need to know if there are any active subscriptions, like this variable.

I have got into the http middleware pipeline, so I can get the websocket requests there. But I am not able to figure out when the websocket connection closes. That is, catch the exception that is supposed to come when a user closes the web browser and abruptly closes the websocket connection.

Does anyone know how to get a list of active subscriptions from GraphQL?

[Shane32]

What are you using for the IObservable? System.Reactive.Subjects.ReplaySubject?

[Shane32]

FYI, if you're adding code into the ASP.NET Core HTTP pipeline, you should have two easy options:

1. use a middleware delegate which has a try/finally block around calling next(options).
2. register a class to be disposed of when the connection is closed - see RegisterForDispose

However, neither of these actually helps determine how many active subscriptions there are, especially important if you have multiple available subscriptions or if you run queries/mutations over WebSocket connections.

I would write a custom IObservable implementation or do something like this (from ChatGPT):

// Original observable
IObservable<int> source;

// Reference count for active subscriptions
int refCount = 0;

// Hook into Subscribe and Dispose with ref counting
IObservable<int> hookedObservable = Observable.Create(observer =>
{
    // Increment the reference count atomically
    if (Interlocked.Increment(ref refCount) == 1)
    {
        // Code to run when the first subscription is made
        Console.WriteLine("First subscription setup");
    }

    Console.WriteLine("Subscribed");

    // Subscribe to the original observable
    try
    {
        var subscription = source.Subscribe(
            observer.OnNext,
            observer.OnError,
            observer.OnCompleted
        );
    
        // Return a disposable that hooks into the Dispose method
        return Disposable.Create(() =>
        {
            Console.WriteLine("Disposed");
    
            // Decrement the reference count atomically
            if (Interlocked.Decrement(ref refCount) == 0)
            {
                // Code to run when the last subscription is disposed
                Console.WriteLine("Last subscription cleaned up");
            }
    
            subscription.Dispose();
        });
    }
    catch
    {
        Console.WriteLine("Failed to set up subscription");

        // Decrement the reference count atomically
        if (Interlocked.Decrement(ref refCount) == 0)
        {
            // Code to run when the last subscription is disposed
            Console.WriteLine("Last subscription cleaned up");
        }

        throw;
    }
});

[KjellOlav]

Thanks @Shane32! You have definitely pointed me in the right direction, although I have some challenges left. ResolveStream in GraphQL wants to have an IObservable, while the more standard observable pattern accepts an IDisposable.

public VariableSubscription(SubscriptionHandler handler)
{ 
    Field<VariableListResult>("variables")
   .Resolve(context => context.Source as VariableList)
   .ResolveStream(context => handler.Subscribe());         
}

And Subscribe returns an IObservable, while a standard best-practice should have returned an IDisposable.

public IObservable<VariableList> Subscribe(List<string> variables, int throttle)
{
    IObservable<VariableList> observable = AddOrGetCollection(variables);
    return observable;
}

This has worked fine for many years, but digging into this now I see that the number of subscribers is just accumulating, they are never unsubscribed. How is unsubscribing from an observable supposed to work? Since ResolveStream doesn't accept an IDisposable and will unsubscribe. What am I missing?

[Shane32]

IObservable contains a single method called Subscribe, which returns an IDisposable. The concept is that the subscriber calls Subscribe, and when it's done with the stream, calls Dispose on the returned IDisposable. GraphQL.NET Server guarantees that the Dispose method will be called when the connection is broken (by either party, expectedly or unexpectedly) or the client issues a 'stop' request for the subscription.

sequenceDiagram
    participant Client
    participant Observable as IObservable
    participant Observer as IObserver
    participant Subscription as IDisposable

    Client->>Observable: Subscribe(Observer)
    Observable->>Client: returns Subscription (IDisposable)
    
    loop Sending Messages
        Observable->>Observer: OnNext(data)
    end
    
    Client->>Subscription: Dispose()
    Subscription->>Observable: Unsubscribe Observer

Loading

This presents at least two ways to set up your data flow inside your application, and another that always causes problems:

1. Events are always broadcast throughout the program using a static IObservable instance.

As such, you can hook into receiving a copy of these (and perhaps perform filtering, etc) for GraphQL clients. Simply use the System.Reactive .Where or .Select (if necessary) and pass the IObservable from the Where/Select to GraphQL. Let's assume you use Where to filter events that pertain to a specific user. The Where method's returned IObservable just contains metadata and does not set up a subscription yet. Then GraphQL calls Subscribe, and System.Reactive internally maps the IObserver and calls the base IObservable, just like LINQ. When OnNext occurs, it filters the request and passes on the data if indicated by the predicate. When Dispose is called, it calls Dispose on the underlying IObservable. Meanwhile, the underlying IObservable keeps track of all connected IObserver instances and copies incoming data events to each connected IObserver, removing the IObserver when Dispose is called on the unique IDisposable interface that it has generated for each IObserver during the call to Subscribe.

Here's a ChatGPT-generated graph of how this might be visualized:

sequenceDiagram
    participant GraphQLClient as GraphQL Client
    participant FilteredObservable as Filtered IObservable (Where)
    participant OriginalObservable as Original IObservable
    participant FilteredObserver as Filtered IObserver
    participant OriginalObserver as Original IObserver
    participant Subscription as IDisposable

    GraphQLClient->>OriginalObservable: Where(FilterPredicate)
    OriginalObservable->>GraphQLClient: returns FilteredObservable (metadata only)
    
    GraphQLClient->>FilteredObservable: Subscribe(FilteredObserver)
    FilteredObservable->>OriginalObservable: Subscribe(OriginalObserver)
    OriginalObservable->>GraphQLClient: returns Subscription (IDisposable)

    loop OnNext Occurs
        OriginalObservable->>OriginalObserver: OnNext(data)
        OriginalObserver->>FilteredObserver: OnNext(data) (if FilterPredicate is true)
        FilteredObserver->>GraphQLClient: OnNext(filtered data)
    end

    GraphQLClient->>Subscription: Dispose()
    Subscription->>OriginalObservable: Unsubscribe(OriginalObserver)

Loading

2. You set up a dedicated event stream per client request.

To do this properly, your IObservable instance should only contain metadata -- whatever information is required to set up the event stream, like the user id of client. It should wait for a call to Subscribe, and then set up the data to flow from the client to the IObserver's OnNext method and provide an IDisposable which will tear down the source event stream. The GraphQL client will call Subscribe when it's ready, and will call Dispose when it's done receiving data.

The wrong way to do it is to set up a stream of data when the IObservable is requested. First, it's difficult to tell when to shut down the connection because IObservable does not inherit from IDisposable. Plus in theory, GraphQL may never call Subscribe, so it would never have an IDisposable to call Dispose on.

[Shane32]

This has worked fine for many years, but digging into this now I see that the number of subscribers is just accumulating, they are never unsubscribed.

Without seeing more of your SubscriptionHandler class, I can't say. In my own codebases, I also sometimes "incorrectly" use the method name of Subscribe to return an IObservable. It is important to note that if it's returning an IObservable, it shouldn't actually be subscribing at all, but merely returning a class containing metadata -- in this case, probably encapsulating variables and/or throttle. Subscription only should occur when Subscribe(IObserver) is called on the returned IObservable, which will return a IDisposable to shut down the subscription.