Merge pull request #240 from graphql-dotnet/fix-websocket-send-queue

Fix websocket send queue

Author: rose-a

src/GraphQL.Client.LocalExecution/GraphQLLocalExecutionClient.cs

@@ -79,8 +79,11 @@ private async Task<GraphQLResponse<TResponse>> ExecuteQueryAsync<TResponse>(Grap
         private async Task<IObservable<GraphQLResponse<TResponse>>> ExecuteSubscriptionAsync<TResponse>(GraphQLRequest request, CancellationToken cancellationToken = default)
         {
             var result = await ExecuteAsync(request, cancellationToken);
-            return ((SubscriptionExecutionResult)result).Streams?.Values.SingleOrDefault()?
-                .SelectMany(executionResult => Observable.FromAsync(token => ExecutionResultToGraphQLResponse<TResponse>(executionResult, token)));
+            var stream = ((SubscriptionExecutionResult)result).Streams?.Values.SingleOrDefault();
+            
+            return stream == null
+                ? Observable.Throw<GraphQLResponse<TResponse>>(new InvalidOperationException("the GraphQL execution did not return an observable"))
+                : stream.SelectMany(executionResult => Observable.FromAsync(token => ExecutionResultToGraphQLResponse<TResponse>(executionResult, token)));
         }
 
         private async Task<ExecutionResult> ExecuteAsync(GraphQLRequest request, CancellationToken cancellationToken = default)

src/GraphQL.Client/GraphQLHttpClient.cs

@@ -2,7 +2,9 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Net.Http;
+using System.Net.Http.Headers;
 using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Client.Abstractions;
@@ -57,6 +59,10 @@ public GraphQLHttpClient(GraphQLHttpClientOptions options, IGraphQLWebsocketJson
             Options = options ?? throw new ArgumentNullException(nameof(options));
             JsonSerializer = serializer ?? throw new ArgumentNullException(nameof(serializer), "please configure the JSON serializer you want to use");
             HttpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
+
+            if (!HttpClient.DefaultRequestHeaders.UserAgent.Any())
+                HttpClient.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue(GetType().Assembly.GetName().Name, GetType().Assembly.GetName().Version.ToString()));
+
             _graphQlHttpWebSocket = new GraphQLHttpWebSocket(GetWebSocketUri(), this);
         }
 

src/GraphQL.Client/Websocket/GraphQLHttpWebSocket.cs

@@ -3,6 +3,7 @@
 using System.IO;
 using System.Net.Http;
 using System.Net.WebSockets;
+using System.Reactive;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
@@ -84,7 +85,8 @@ public GraphQLHttpWebSocket(Uri webSocketUri, GraphQLHttpClient client)
 
             _requestSubscription = _requestSubject
                 .ObserveOn(_sendLoopScheduler)
-                .Subscribe(async request => await SendWebSocketRequestAsync(request));
+                .SelectMany(SendWebSocketRequestAsync)
+                .Subscribe();
         }
 
         #region Send requests
@@ -339,14 +341,14 @@ private Task QueueWebSocketRequest(GraphQLWebSocketRequest request)
             return request.SendTask();
         }
 
-        private async Task SendWebSocketRequestAsync(GraphQLWebSocketRequest request)
+        private async Task<Unit> SendWebSocketRequestAsync(GraphQLWebSocketRequest request)
         {
             try
             {
                 if (_internalCancellationToken.IsCancellationRequested)
                 {
                     request.SendCanceled();
-                    return;
+                    return Unit.Default;
                 }
 
                 await InitializeWebSocket();
@@ -362,6 +364,7 @@ await _clientWebSocket.SendAsync(
             {
                 request.SendFailed(e);
             }
+            return Unit.Default;
         }
 
         #endregion

src/src.props

@@ -7,9 +7,8 @@
     <LangVersion>8.0</LangVersion>
   </PropertyGroup>
 
-  <!--TODO: remove condition when https://github.com/GitTools/GitVersion/issues/2063 is fixed-->
-  <ItemGroup Condition=" '$(DEBUG)' != ''">
-    <PackageReference Update="GitVersionTask" Version="5.2.4">
+  <ItemGroup>
+    <PackageReference Update="GitVersionTask" Version="5.3.4">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>