Skip to content

Commit 477b7f4

Browse files
rose-arose-a
authored
Merge pull request #368 from graphql-dotnet/improve-di-for-local-execution-client
Fix request preprocessing for full websocket transport
2 parents 1116d2b + 7cb148e commit 477b7f4

File tree

4 files changed

+85
-68
lines changed

4 files changed

+85
-68
lines changed

‎src/GraphQL.Client.LocalExecution/GraphQL.Client.LocalExecution.csproj‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,9 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="GraphQL" Version="4.6.0" />
11+
<PackageReference Include="GraphQL.MicrosoftDI" Version="4.6.1" />
1212
<PackageReference Include="GraphQL.NewtonsoftJson" Version="4.6.0" />
13+
<PackageReference Include="GraphQL.SystemReactive" Version="4.1.0" />
1314
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
1415
</ItemGroup>
1516

‎src/GraphQL.Client.LocalExecution/GraphQLLocalExecutionClient.cs‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@
44
using System.Linq;
55
using System.Reactive.Linq;
66
using System.Reactive.Threading.Tasks;
7-
using System.Text;
87
using System.Threading;
98
using System.Threading.Tasks;
109
using GraphQL.Client.Abstractions;
@@ -21,7 +20,7 @@ namespace GraphQL.Client.LocalExecution
2120
public static class GraphQLLocalExecutionClient
2221
{
2322
public static GraphQLLocalExecutionClient<TSchema> New<TSchema>(TSchema schema, IGraphQLJsonSerializer serializer) where TSchema : ISchema
24-
=> new GraphQLLocalExecutionClient<TSchema>(schema, serializer);
23+
=> new GraphQLLocalExecutionClient<TSchema>(schema, serializer, new SubscriptionDocumentExecuter(), new DocumentWriter());
2524
}
2625

2726
public class GraphQLLocalExecutionClient<TSchema> : IGraphQLClient where TSchema : ISchema
@@ -41,18 +40,18 @@ public class GraphQLLocalExecutionClient<TSchema> : IGraphQLClient where TSchema
4140

4241
public IGraphQLJsonSerializer Serializer { get; }
4342

44-
private readonly DocumentExecuter _documentExecuter;
45-
private readonly DocumentWriter _documentWriter;
43+
private readonly IDocumentExecuter _documentExecuter;
44+
private readonly IDocumentWriter _documentWriter;
4645

47-
public GraphQLLocalExecutionClient(TSchema schema, IGraphQLJsonSerializer serializer)
46+
public GraphQLLocalExecutionClient(TSchema schema, IGraphQLJsonSerializer serializer, IDocumentExecuter documentExecuter, IDocumentWriter documentWriter)
4847
{
4948
Schema = schema ?? throw new ArgumentNullException(nameof(schema), "no schema configured");
5049
Serializer = serializer ?? throw new ArgumentNullException(nameof(serializer), "please configure the JSON serializer you want to use");
5150

5251
if (!Schema.Initialized)
5352
Schema.Initialize();
54-
_documentExecuter = new DocumentExecuter();
55-
_documentWriter = new DocumentWriter();
53+
_documentExecuter = documentExecuter;
54+
_documentWriter = documentWriter;
5655
}
5756

5857
public void Dispose() { }
@@ -78,7 +77,7 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
7877
private async Task<GraphQLResponse<TResponse>> ExecuteQueryAsync<TResponse>(GraphQLRequest request, CancellationToken cancellationToken)
7978
{
8079
var executionResult = await ExecuteAsync(request, cancellationToken);
81-
return await ExecutionResultToGraphQLResponse<TResponse>(executionResult, cancellationToken);
80+
return await ExecutionResultToGraphQLResponseAsync<TResponse>(executionResult, cancellationToken);
8281
}
8382
private async Task<IObservable<GraphQLResponse<TResponse>>> ExecuteSubscriptionAsync<TResponse>(GraphQLRequest request, CancellationToken cancellationToken = default)
8483
{
@@ -87,12 +86,12 @@ private async Task<IObservable<GraphQLResponse<TResponse>>> ExecuteSubscriptionA
8786

8887
return stream == null
8988
? Observable.Throw<GraphQLResponse<TResponse>>(new InvalidOperationException("the GraphQL execution did not return an observable"))
90-
: stream.SelectMany(executionResult => Observable.FromAsync(token => ExecutionResultToGraphQLResponse<TResponse>(executionResult, token)));
89+
: stream.SelectMany(executionResult => Observable.FromAsync(token => ExecutionResultToGraphQLResponseAsync<TResponse>(executionResult, token)));
9190
}
9291

9392
private async Task<ExecutionResult> ExecuteAsync(GraphQLRequest request, CancellationToken cancellationToken = default)
9493
{
95-
var serializedRequest = Serializer.SerializeToString(request);
94+
string serializedRequest = Serializer.SerializeToString(request);
9695

9796
var deserializedRequest = JsonConvert.DeserializeObject<GraphQLRequest>(serializedRequest);
9897
var inputs = deserializedRequest.Variables != null
@@ -103,22 +102,21 @@ private async Task<ExecutionResult> ExecuteAsync(GraphQLRequest request, Cancell
103102
var result = await _documentExecuter.ExecuteAsync(options =>
104103
{
105104
options.Schema = Schema;
106-
options.OperationName = request.OperationName;
107-
options.Query = request.Query;
105+
options.OperationName = deserializedRequest?.OperationName;
106+
options.Query = deserializedRequest?.Query;
108107
options.Inputs = inputs;
109108
options.CancellationToken = cancellationToken;
110109
});
111110

112111
return result;
113112
}
114113

115-
private async Task<GraphQLResponse<TResponse>> ExecutionResultToGraphQLResponse<TResponse>(ExecutionResult executionResult, CancellationToken cancellationToken = default)
114+
private async Task<GraphQLResponse<TResponse>> ExecutionResultToGraphQLResponseAsync<TResponse>(ExecutionResult executionResult, CancellationToken cancellationToken = default)
116115
{
117-
string json = await _documentWriter.WriteToStringAsync(executionResult);
118-
// serialize result into utf8 byte stream
119-
var resultStream = new MemoryStream(Encoding.UTF8.GetBytes(json));
120-
// deserialize using the provided serializer
121-
return await Serializer.DeserializeFromUtf8StreamAsync<TResponse>(resultStream, cancellationToken);
116+
using var stream = new MemoryStream();
117+
await _documentWriter.WriteAsync(stream, executionResult, cancellationToken);
118+
stream.Seek(0, SeekOrigin.Begin);
119+
return await Serializer.DeserializeFromUtf8StreamAsync<TResponse>(stream, cancellationToken);
122120
}
123121

124122
#endregion

‎src/GraphQL.Client.LocalExecution/ServiceCollectionExtensions.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
using GraphQL.Client.Abstractions;
2+
using GraphQL.DI;
3+
using GraphQL.MicrosoftDI;
4+
using GraphQL.Types;
5+
using Microsoft.Extensions.DependencyInjection;
6+
7+
namespace GraphQL.Client.LocalExecution
8+
{
9+
public static class ServiceCollectionExtensions
10+
{
11+
public static IGraphQLBuilder AddGraphQLLocalExecutionClient<TSchema>(this IServiceCollection services) where TSchema : ISchema
12+
{
13+
services.AddSingleton<GraphQLLocalExecutionClient<TSchema>>();
14+
services.AddSingleton<IGraphQLClient, GraphQLLocalExecutionClient<TSchema>>();
15+
return services.AddGraphQL();
16+
}
17+
}
18+
}

‎src/GraphQL.Client/Websocket/GraphQLHttpWebSocket.cs‎

Lines changed: 49 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,7 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
182182
catch (OperationCanceledException) { }
183183
})
184184
);
185-
185+
186186
Debug.WriteLine($"sending start message on subscription {startRequest.Id}");
187187
// send subscription request
188188
try
@@ -265,55 +265,55 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
265265
/// <returns></returns>
266266
public Task<GraphQLResponse<TResponse>> SendRequest<TResponse>(GraphQLRequest request, CancellationToken cancellationToken = default) =>
267267
Observable.Create<GraphQLResponse<TResponse>>(async observer =>
268+
{
269+
var preprocessedRequest = await _client.Options.PreprocessRequest(request, _client);
270+
var websocketRequest = new GraphQLWebSocketRequest
268271
{
269-
await _client.Options.PreprocessRequest(request, _client);
270-
var websocketRequest = new GraphQLWebSocketRequest
272+
Id = Guid.NewGuid().ToString("N"),
273+
Type = GraphQLWebSocketMessageType.GQL_START,
274+
Payload = preprocessedRequest
275+
};
276+
var observable = IncomingMessageStream
277+
.Where(response => response != null && response.Id == websocketRequest.Id)
278+
.TakeUntil(response => response.Type == GraphQLWebSocketMessageType.GQL_COMPLETE)
279+
.Select(response =>
271280
{
272-
Id = Guid.NewGuid().ToString("N"),
273-
Type = GraphQLWebSocketMessageType.GQL_START,
274-
Payload = request
275-
};
276-
var observable = IncomingMessageStream
277-
.Where(response => response != null && response.Id == websocketRequest.Id)
278-
.TakeUntil(response => response.Type == GraphQLWebSocketMessageType.GQL_COMPLETE)
279-
.Select(response =>
280-
{
281-
Debug.WriteLine($"received response for request {websocketRequest.Id}");
282-
var typedResponse =
283-
_client.JsonSerializer.DeserializeToWebsocketResponse<TResponse>(
284-
response.MessageBytes);
285-
return typedResponse.Payload;
286-
});
281+
Debug.WriteLine($"received response for request {websocketRequest.Id}");
282+
var typedResponse =
283+
_client.JsonSerializer.DeserializeToWebsocketResponse<TResponse>(
284+
response.MessageBytes);
285+
return typedResponse.Payload;
286+
});
287287

288-
try
289-
{
290-
// initialize websocket (completes immediately if socket is already open)
291-
await InitializeWebSocket();
292-
}
293-
catch (Exception e)
294-
{
295-
// subscribe observer to failed observable
296-
return Observable.Throw<GraphQLResponse<TResponse>>(e).Subscribe(observer);
297-
}
288+
try
289+
{
290+
// initialize websocket (completes immediately if socket is already open)
291+
await InitializeWebSocket();
292+
}
293+
catch (Exception e)
294+
{
295+
// subscribe observer to failed observable
296+
return Observable.Throw<GraphQLResponse<TResponse>>(e).Subscribe(observer);
297+
}
298298

299-
var disposable = new CompositeDisposable(
300-
observable.Subscribe(observer)
301-
);
299+
var disposable = new CompositeDisposable(
300+
observable.Subscribe(observer)
301+
);
302302

303-
Debug.WriteLine($"submitting request {websocketRequest.Id}");
304-
// send request
305-
try
306-
{
307-
await QueueWebSocketRequest(websocketRequest);
308-
}
309-
catch (Exception e)
310-
{
311-
Debug.WriteLine(e);
312-
throw;
313-
}
303+
Debug.WriteLine($"submitting request {websocketRequest.Id}");
304+
// send request
305+
try
306+
{
307+
await QueueWebSocketRequest(websocketRequest);
308+
}
309+
catch (Exception e)
310+
{
311+
Debug.WriteLine(e);
312+
throw;
313+
}
314314

315-
return disposable;
316-
})
315+
return disposable;
316+
})
317317
// complete sequence on OperationCanceledException, this is triggered by the cancellation token
318318
.Catch<GraphQLResponse<TResponse>, OperationCanceledException>(exception =>
319319
Observable.Empty<GraphQLResponse<TResponse>>())
@@ -410,7 +410,7 @@ public Task InitializeWebSocket()
410410
}
411411
catch (NotImplementedException)
412412
{
413-
Debug.WriteLine("property 'ClientWebSocketOptions.ClientCertificates' not implemented by current platform");
413+
Debug.WriteLine("property 'ClientWebSocketOptions.ClientCertificates' not implemented by current platform");
414414
}
415415
catch (PlatformNotSupportedException)
416416
{
@@ -423,7 +423,7 @@ public Task InitializeWebSocket()
423423
}
424424
catch (NotImplementedException)
425425
{
426-
Debug.WriteLine("property 'ClientWebSocketOptions.UseDefaultCredentials' not implemented by current platform");
426+
Debug.WriteLine("property 'ClientWebSocketOptions.UseDefaultCredentials' not implemented by current platform");
427427
}
428428
catch (PlatformNotSupportedException)
429429
{
@@ -479,7 +479,7 @@ private async Task ConnectAsync(CancellationToken token)
479479
Debug.WriteLine($"new incoming message stream {_incomingMessages.GetHashCode()} created");
480480

481481
_incomingMessagesConnection = new CompositeDisposable(maintenanceSubscription, connection);
482-
482+
483483
var initRequest = new GraphQLWebSocketRequest
484484
{
485485
Type = GraphQLWebSocketMessageType.GQL_CONNECTION_INIT,
@@ -488,7 +488,7 @@ private async Task ConnectAsync(CancellationToken token)
488488

489489
// setup task to await connection_ack message
490490
var ackTask = _incomingMessages
491-
.Where(response => response != null )
491+
.Where(response => response != null)
492492
.TakeUntil(response => response.Type == GraphQLWebSocketMessageType.GQL_CONNECTION_ACK ||
493493
response.Type == GraphQLWebSocketMessageType.GQL_CONNECTION_ERROR)
494494
.LastAsync()
@@ -640,7 +640,7 @@ private async Task CloseAsync()
640640
}
641641

642642
Debug.WriteLine($"send \"connection_terminate\" message");
643-
await SendWebSocketMessageAsync(new GraphQLWebSocketRequest{Type = GraphQLWebSocketMessageType.GQL_CONNECTION_TERMINATE});
643+
await SendWebSocketMessageAsync(new GraphQLWebSocketRequest { Type = GraphQLWebSocketMessageType.GQL_CONNECTION_TERMINATE });
644644

645645
Debug.WriteLine($"closing websocket {_clientWebSocket.GetHashCode()}");
646646
await _clientWebSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None);

0 commit comments

Comments
 (0)