Support graphql-transport-ws websocket protocol (#539)

* Preparing code to add graphql-transport-ws protocol without breaking previous dependant code

* Added graphql-transport-ws subprotocol. Tested with Graphql Yoga server

* Renamed protocols (dropping the deprecated keyword). Split GraphQLHttpWebSocket into two classes (one per protocol) that inherit from BaseGraphQLHttpWebSocket

* fix formatting, do some refactoring

* create unit tests for graphql-transport-ws protocol

* catch json exceptions on empty close messages

* fix handling of regular requests and errors

* change IGraphQLWebsocketSerializer to support diffenent payload types +semver: breaking

* properly hook up ping pong

* implement sub protocol auto-negotiation

* implement and test ping/ping

* fix formatting error

---------

Co-authored-by: joao-avelino <[email protected]>

Author: rose-a

src/GraphQL.Client.Abstractions.Websocket/GraphQLWebSocketMessageType.cs

@@ -72,15 +72,67 @@ public static class GraphQLWebSocketMessageType
     public const string GQL_ERROR = "error"; // Server -> Client
 
     /// <summary>
-    ///     Server sends this message to indicate that a GraphQL operation is done, and no more data will arrive for the
-    ///     specific operation.
+    ///     Server -> Client: Server sends this message to indicate that a GraphQL operation is done, and no more data will arrive
+    ///     for the specific operation.
+    ///     Client -> Server: "indicates that the client has stopped listening and wants to complete the subscription. No further
+    ///     events, relevant to the original subscription, should be sent through. Even if the client sent a Complete message for
+    ///     a single-result-operation before it resolved, the result should not be sent through once it does."
+    ///     Replaces the GQL_STOP in graphql-transport-ws
     ///     id: string : operation ID of the operation that completed
     /// </summary>
-    public const string GQL_COMPLETE = "complete"; // Server -> Client
+    public const string GQL_COMPLETE = "complete"; // Server -> Client and Client -> Server
 
     /// <summary>
     ///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
     ///     id: string : operation id
     /// </summary>
     public const string GQL_STOP = "stop"; // Client -> Server
+
+
+    // Additional types for graphql-transport-ws, as described in https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md
+
+    /// <summary>
+    ///     Bidirectional. "Useful for detecting failed connections, displaying latency metrics or other types of network probing.
+    ///     A Pong must be sent in response from the receiving party as soon as possible. The Ping message can be sent at any time
+    ///     within the established socket. The optional payload field can be used to transfer additional details about the ping."
+    ///     payload: Object: ping details
+    /// </summary>
+    public const string GQL_PING = "ping"; // Bidirectional
+
+    /// <summary>
+    ///     Bidirectional. "The response to the Ping message. Must be sent as soon as the Ping message is received.
+    ///     The Pong message can be sent at any time within the established socket. Furthermore, the Pong message
+    ///     may even be sent unsolicited as an unidirectional heartbeat. The optional payload field can be used to
+    ///     transfer additional details about the pong."
+    ///     payload: Object: pong details
+    /// </summary>
+    public const string GQL_PONG = "pong"; // Bidirectional
+
+    /// <summary>
+    ///     Client-> Server. "Requests an operation specified in the message payload. This message provides a unique
+    ///     ID field to connect published messages to the operation requested by this message. If there is already an
+    ///     active subscriber for an operation matching the provided ID, regardless of the operation type, the server
+    ///     must close the socket immediately with the event 4409: Subscriber for *unique-operation-id* already exists.
+    ///     The server needs only keep track of IDs for as long as the subscription is active. Once a client completes
+    ///     an operation, it is free to re-use that ID."
+    ///     id: string : operation id
+    ///     payload: Object:
+    ///         operationName : string : subscribe
+    ///         query : string : the subscription query
+    ///         variables : Dictionary(string, string) : a dictionary with variables and their values
+    ///         extensions : Dictionary(string, string) : a dictionary of extensions
+    /// </summary>
+    public const string GQL_SUBSCRIBE = "subscribe"; // Client -> Server
+
+
+    /// <summary>
+    ///     Server -> Client. "Operation execution result(s) from the source stream created by the binding Subscribe
+    ///     message. After all results have been emitted, the Complete message will follow indicating stream
+    ///     completion."
+    ///     id: string : operation id
+    ///     payload: Object: ExecutionResult
+    /// </summary>
+
+    public const string GQL_NEXT = "next"; // Server -> Client
+    
 }

src/GraphQL.Client.Abstractions.Websocket/IGraphQLWebSocketClient.cs

@@ -2,6 +2,11 @@ namespace GraphQL.Client.Abstractions.Websocket;
 
 public interface IGraphQLWebSocketClient : IGraphQLClient
 {
+    /// <summary>
+    /// The negotiated websocket sub-protocol. Will be <see langword="null"/> while no websocket connection is established.
+    /// </summary>
+    string? WebSocketSubProtocol { get; }
+
     /// <summary>
     /// Publishes all exceptions which occur inside the websocket receive stream (i.e. for logging purposes)
     /// </summary>
@@ -16,4 +21,25 @@ public interface IGraphQLWebSocketClient : IGraphQLClient
     /// Explicitly opens the websocket connection. Will be closed again on disposing the last subscription.
     /// </summary>
     Task InitializeWebsocketConnection();
+
+    /// <summary>
+    /// Publishes the payload of all received pong messages (which may be <see langword="null"/>). Subscribing initiates the websocket connection. <br/>
+    /// Ping/Pong is only supported when using the "graphql-transport-ws" websocket sub-protocol.
+    /// </summary>
+    /// <exception cref="NotSupportedException">the negotiated websocket sub-protocol does not support ping/pong</exception>
+    IObservable<object?> PongStream { get; }
+
+    /// <summary>
+    /// Sends a ping to the server. <br/>
+    /// Ping/Pong is only supported when using the "graphql-transport-ws" websocket sub-protocol.
+    /// </summary>
+    /// <exception cref="NotSupportedException">the negotiated websocket sub-protocol does not support ping/pong</exception>
+    Task SendPingAsync(object? payload);
+
+    /// <summary>
+    /// Sends a pong to the server. This can be used for keep-alive scenarios (the client will automatically respond to pings received from the server). <br/>
+    /// Ping/Pong is only supported when using the "graphql-transport-ws" websocket sub-protocol.
+    /// </summary>
+    /// <exception cref="NotSupportedException">the negotiated websocket sub-protocol does not support ping/pong</exception>
+    Task SendPongAsync(object? payload);
 }

src/GraphQL.Client.Abstractions.Websocket/IGraphQLWebsocketJsonSerializer.cs

@@ -10,5 +10,5 @@ public interface IGraphQLWebsocketJsonSerializer : IGraphQLJsonSerializer
 
     Task<WebsocketMessageWrapper> DeserializeToWebsocketResponseWrapperAsync(Stream stream);
 
-    GraphQLWebSocketResponse<GraphQLResponse<TResponse>> DeserializeToWebsocketResponse<TResponse>(byte[] bytes);
+    GraphQLWebSocketResponse<TResponse> DeserializeToWebsocketResponse<TResponse>(byte[] bytes);
 }

src/GraphQL.Client.Serializer.Newtonsoft/NewtonsoftJsonSerializer.cs

@@ -40,8 +40,8 @@ public byte[] SerializeToBytes(GraphQLWebSocketRequest request)
 
     public Task<WebsocketMessageWrapper> DeserializeToWebsocketResponseWrapperAsync(Stream stream) => DeserializeFromUtf8Stream<WebsocketMessageWrapper>(stream);
 
-    public GraphQLWebSocketResponse<GraphQLResponse<TResponse>> DeserializeToWebsocketResponse<TResponse>(byte[] bytes) =>
-        JsonConvert.DeserializeObject<GraphQLWebSocketResponse<GraphQLResponse<TResponse>>>(Encoding.UTF8.GetString(bytes),
+    public GraphQLWebSocketResponse<TResponse> DeserializeToWebsocketResponse<TResponse>(byte[] bytes) =>
+        JsonConvert.DeserializeObject<GraphQLWebSocketResponse<TResponse>>(Encoding.UTF8.GetString(bytes),
             JsonSerializerSettings);
 
     public Task<GraphQLResponse<TResponse>> DeserializeFromUtf8StreamAsync<TResponse>(Stream stream, CancellationToken cancellationToken) => DeserializeFromUtf8Stream<GraphQLResponse<TResponse>>(stream);

src/GraphQL.Client.Serializer.SystemTextJson/SystemTextJsonSerializer.cs

@@ -42,7 +42,7 @@ private void ConfigureMandatorySerializerOptions()
 
     public Task<WebsocketMessageWrapper> DeserializeToWebsocketResponseWrapperAsync(Stream stream) => JsonSerializer.DeserializeAsync<WebsocketMessageWrapper>(stream, Options).AsTask();
 
-    public GraphQLWebSocketResponse<GraphQLResponse<TResponse>> DeserializeToWebsocketResponse<TResponse>(byte[] bytes) =>
-        JsonSerializer.Deserialize<GraphQLWebSocketResponse<GraphQLResponse<TResponse>>>(new ReadOnlySpan<byte>(bytes),
+    public GraphQLWebSocketResponse<TResponse> DeserializeToWebsocketResponse<TResponse>(byte[] bytes) =>
+        JsonSerializer.Deserialize<GraphQLWebSocketResponse<TResponse>>(new ReadOnlySpan<byte>(bytes),
             Options);
 }

src/GraphQL.Client/GraphQLHttpClient.cs

@@ -30,16 +30,18 @@ public class GraphQLHttpClient : IGraphQLWebSocketClient, IDisposable
     /// </summary>
     public GraphQLHttpClientOptions Options { get; }
 
-    /// <summary>
-    /// Publishes all exceptions which occur inside the websocket receive stream (i.e. for logging purposes)
-    /// </summary>
+    /// <inheritdoc />
     public IObservable<Exception> WebSocketReceiveErrors => GraphQlHttpWebSocket.ReceiveErrors;
 
-    /// <summary>
-    /// the websocket connection state
-    /// </summary>
+    /// <inheritdoc />
+    public string? WebSocketSubProtocol => GraphQlHttpWebSocket.WebsocketProtocol;
+
+    /// <inheritdoc />
     public IObservable<GraphQLWebsocketConnectionState> WebsocketConnectionState => GraphQlHttpWebSocket.ConnectionState;
 
+    /// <inheritdoc />
+    public IObservable<object?> PongStream => GraphQlHttpWebSocket.GetPongStream();
+
     #region Constructors
 
     public GraphQLHttpClient(string endPoint, IGraphQLWebsocketJsonSerializer serializer)
@@ -78,7 +80,7 @@ public GraphQLHttpClient(GraphQLHttpClientOptions options, IGraphQLWebsocketJson
     public async Task<GraphQLResponse<TResponse>> SendQueryAsync<TResponse>(GraphQLRequest request, CancellationToken cancellationToken = default)
     {
         return Options.UseWebSocketForQueriesAndMutations || Options.WebSocketEndPoint is not null && Options.EndPoint is null || Options.EndPoint.HasWebSocketScheme()
-            ? await GraphQlHttpWebSocket.SendRequest<TResponse>(request, cancellationToken).ConfigureAwait(false)
+            ? await GraphQlHttpWebSocket.SendRequestAsync<TResponse>(request, cancellationToken).ConfigureAwait(false)
             : await SendHttpRequestAsync<TResponse>(request, cancellationToken).ConfigureAwait(false);
     }
 
@@ -103,12 +105,15 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
 
     #endregion
 
-    /// <summary>
-    /// Explicitly opens the websocket connection. Will be closed again on disposing the last subscription.
-    /// </summary>
-    /// <returns></returns>
+    /// <inheritdoc />
     public Task InitializeWebsocketConnection() => GraphQlHttpWebSocket.InitializeWebSocket();
 
+    /// <inheritdoc />
+    public Task SendPingAsync(object? payload) => GraphQlHttpWebSocket.SendPingAsync(payload);
+
+    /// <inheritdoc />
+    public Task SendPongAsync(object? payload) => GraphQlHttpWebSocket.SendPongAsync(payload);
+
     #region Private Methods
 
     private async Task<GraphQLHttpResponse<TResponse>> SendHttpRequestAsync<TResponse>(GraphQLRequest request, CancellationToken cancellationToken = default)
@@ -143,9 +148,9 @@ private GraphQLHttpWebSocket CreateGraphQLHttpWebSocket()
             throw new InvalidOperationException("no endpoint configured");
 
         var webSocketEndpoint = Options.WebSocketEndPoint ?? Options.EndPoint.GetWebSocketUri();
-        return webSocketEndpoint.HasWebSocketScheme()
-            ? new GraphQLHttpWebSocket(webSocketEndpoint, this)
-            : throw new InvalidOperationException($"uri \"{webSocketEndpoint}\" is not a websocket endpoint");
+        return !webSocketEndpoint.HasWebSocketScheme()
+            ? throw new InvalidOperationException($"uri \"{webSocketEndpoint}\" is not a websocket endpoint")
+            : new GraphQLHttpWebSocket(webSocketEndpoint, this);
     }
 
     #endregion

src/GraphQL.Client/GraphQLHttpClientOptions.cs

@@ -1,6 +1,7 @@
 using System.Net;
 using System.Net.Http.Headers;
 using System.Net.WebSockets;
+using GraphQL.Client.Http.Websocket;
 
 namespace GraphQL.Client.Http;
 
@@ -19,6 +20,11 @@ public class GraphQLHttpClientOptions
     /// </summary>
     public Uri? WebSocketEndPoint { get; set; }
 
+    /// <summary>
+    /// The GraphQL websocket protocol to be used. Defaults to the older "graphql-ws" protocol to not break old code. 
+    /// </summary>
+    public string? WebSocketProtocol { get; set; } = WebSocketProtocols.AUTO_NEGOTIATE;
+
     /// <summary>
     /// The <see cref="System.Net.Http.HttpMessageHandler"/> that is going to be used
     /// </summary>