Proposal: GraphQL client subscription support

[MohamedSabthar]

Summary

The ballerina graphql package provides client implementation to execute query and mutation operations by connecting to the server. But the subscription operation is not supported by the client. This proposal is to add subscription support to the existing client implementation.

Goals

• Support subscription operation with graphql:Client

Motivation

Ballerina doesn’t provide a simpler way to execute subscription operations by connecting to a graphql service. This is currently handled manually by creating a WebSocket client and writing the graphql subprotocols (graphql-ws/graphql-transport-ws) related code. Having the subscription support in the graphql client will reduce writing boilerplate subprotocol-related code and allow users to interact with the services with simple API.

Description

Similar to query operations, subscription operations enable the user to fetch data. Unlike queries, subscriptions are long-lasting operations that can change their result over time. They can maintain an active connection to the GraphQL server (most commonly via WebSocket), enabling the server to push updates to the subscription's result. This subscription operation is implemented in the ballerina graphql service using the graphql-ws subprotocol specification; the same protocol will be used to implement subscription operation on the client side. The following section describes the API changes to the existing graphql:Client to incorporate subscription operation.

API changes

ClientConfiguration changes

public type ClientConfiguration record {|
    http:ClientConfiguration httpConfig = {};
    websocket:ClientConfiguration websocketConfig =  {};
|};

The ClientConfiguration is updated to have both HTTP configuration and WebSocket configuration.

Client changes

public isolated client class Client {
  private websocket:Client? wsClient;
  private map<Subscriber> subscribers;
  private websocket:ClientConfiguration wsConfig;
  private string wsServiceUrl;

   … // other fields and methods
}

The client keeps a wsClient as its field and utilizes it when executing subscription operations. The client stores the id(specified in the spec) to Subscriber mapping in a ballerina map called subscribers. This map allows the client to handle multiplexing between subscription responses. More details regarding the Subscriber and the generation of unique id are described in later sections of this proposal.

Apart from the above-described fields, the client keeps the websocketConfig and wsServiceUrl fields. These two fields later will be used by the client to lazily initialize the wsClient when there is a subscription operation executed by the client.

Client init method

The init method of the graphql:Client sets the wsConfig and wsServiceUrl fields. Values for these fields are obtained from init method parameters. For example, the serviceUrl passed in the init method will be programmatically changed to have ws(s) protocol and set as the value of the wsServiceUrl field.

Client execute method

No changes are made to the execute method signature. But the data binding type of this method now includes a new type called Subscriber. Following is the execute method definition after the new type introduction.

remote isolated function execute(string document,
map<anydata>? variables = (),
string? operationName = (),
map<string|string[]>? headers = (),
typedesc<GenericResponseWithErrors|record{}|json|**Subscriber**> targetType = <>)
returns targetType|ClientError = @java:Method {
        'class: "io.ballerina.stdlib.graphql.runtime.client.QueryExecutor",
        name: "execute"
    } external;

If the data binding type is Subscriber the client knows that it’s going to execute a subscription operation. The client generates a new UUID and stores the subscriber instance in the subscribers map whenever it executes a subscription operation. Further, the client checks the wsClient field value, if it is set to nil the wsClient will be initialized to a websocket:Client. The websocketConfig and wsServiceUrl will be used during thiswsClient initialization.

Subscriber class

Subscriber is a client class which holds the stream and id of a particular subscription operation. The client class definition is given below

distinct isolated client class Subscriber {
    private final string id;

    isolated function init(string id) {
        self.id = id;
    }

    isolated remote function unsubscribe() returns websocket:Error? {
        // logic to send ws_complete message along with id
    }

    isolated function getStream(
    typedesc<GenericResponseWithErrors|record{}|json> targetType = <>) 
    returns stream<targetType>|ClientError = @java:Method {
        'class: "io.ballerina.stdlib.graphql.runtime.client.Subscriber",
        name: "getStream"
    } external;
}

After executing the subscription operation the resulting stream can be obtained by executing the getStream method on the Subscriber instance. This getStream method allows data binding on the steam type. Further, the subscription can be cancelled by calling the unsubscribe method on the Subscriber instance.

Client close method

A new close method could be introduced in the graphql:Client. Executing this method sends a “complete/stop” message to all subscription operations and closes the websocket connection. executing subscription operation after calling close() method will return a graphql:ClientError(Websocket connection already closed)

Following is an example showing a client executing a subscription operation.

import ballerina/io;
import ballerina/graphql;

public function main() returns error? {
    // initialize client - this internally initialize a websocket client with ws://localhost:4000/graphql endpoint
    graphql:Client ep = check new ("http://localhost:4000//graphql", websocketConfig = {});

    string document = string `subscription { totalDonations }`;
    // execute subscription operation
    Subscriber subscriber = check ep->execute(document);

    // get stream from subscriber
    stream<Response> responseStream = check subscriber.getStream();

    // process stream
    responseStream.forEach(r => io:println(r.totalDonations));

    // unsubscribe from subscription, server will not send any data after this
    check subscriber->unsubscribe();

    // close the websocket connection
    ep->close();
}

Handling multiple subscriptions with graphql:Client

Following pseudocode explains the client multiplexing logic

• The client generates a unique UUID for each subscription request.
• The client stores the UUID to Subscriber instance mapping in a ballerina subscriber map.
• Foreach a new message from the WebSocket
  • If the message type is next or error:
    • get the payload form the message
    • get the UUID from id field of the message
    • get the subscriber instance from subscriber map using UUID
    • adds the payload of the message to the subscriber results stream
  • else if the message type is complete:
    • remove the subscriber from map
  • else:
    • handle other message types (ex: pong)

[MohamedSabthar]

Blocked on https://github.com/ballerina-platform/ballerina-standard-library/issues/3962

[shafreenAnfar]

Good proposal.

I am thinking if it should be Subscriber or Subscription. I like the later. We can also consider ResponseStream as mentioned in the spec. In that case, unsubscribe should be cancel.

Also, I wonder if close should be, unsubscribeAll assuming unsubscribe closes the socket connection as well.

[shafreenAnfar]

Ballerina streams has a close method in that case do we need this intermediate client object.

[MohamedSabthar]

The following changes are being introduced to the proposal, as discussed with @shafreenAnfar and @ThisaruGuruge:

1. Instead of exposing an intermediate Subscriber client object, the function execute returns a stream.

   remote isolated function execute(string document,
   map<anydata>? variables = (),
   string? operationName = (),
   map<string|string[]>? headers = (),
   typedesc<GenericResponseWithErrors|record{}|json|**stream**> targetType = <>)
   returns targetType|ClientError = @java:Method {
       'class: "io.ballerina.stdlib.graphql.runtime.client.QueryExecutor",
       name: "execute"
   } external;

2. The unsubscribe method of the Subscriber client can be implemented with the stream.close() method.
3. The close method of graphql:Client has been renamed to closeSubscriptions.

following is an example after above updates

import ballerina/io;
import ballerina/graphql;

public function main() returns error? {
    // Initialize client - this internally initialize a websocket client with ws://localhost:4000/graphql endpoint
    graphql:Client ep = check new ("http://localhost:4000//graphql", websocketConfig = {});

    string document = string `subscription { totalDonations }`;
    // execute operation from the return type binding it knows the operation in subscription (if type is stream then it's a subscription)
    stream<Response> responseStream = check ep->execute(document);

    // process stream
    responseStream.forEach(r => io:println(r.totalDonations));

    // close the stream which also send graphql-ws `complete` messsage to the service for this operation
    check responseStream.close();

    // close the websocket connection
    ep->closeSubscriptions();
}

[ThisaruGuruge]

This is deferred due to other priorities.

[MaryamZi]

Is there a specific reason to not have three separate methods for the three operations (e.g., query, mutate, and subscribe) instead of or in addition to execute? With the addition of subscription support, doesn't the return type get a bit complicated, especially given that the return type depends on the type of operation?