Added GraphQL SSE subgraph support to fusion. (#6356)

Author: michaelstaib

src/HotChocolate/AspNetCore/src/Transport.Http/GraphQLHttpEventStreamProcessor.cs

@@ -116,6 +116,7 @@ private static async IAsyncEnumerable<OperationResult> ReadMessagesPipeAsync(
                             yield break;
                         }
 
+                        message.Reset();
                         yield return operationResult;
                     }
 

src/HotChocolate/Fusion/src/Core/Clients/DefaultHttpGraphQLSubscriptionClient.cs

@@ -0,0 +1,57 @@
+using System.Runtime.CompilerServices;
+using HotChocolate.Fusion.Metadata;
+using HotChocolate.Transport.Http;
+
+namespace HotChocolate.Fusion.Clients;
+
+internal sealed class DefaultHttpGraphQLSubscriptionClient : IGraphQLSubscriptionClient
+{
+    private readonly HttpClientConfiguration _config;
+    private readonly DefaultGraphQLHttpClient _client;
+
+    public DefaultHttpGraphQLSubscriptionClient(
+        HttpClientConfiguration configuration,
+        HttpClient httpClient)
+    {
+        if (httpClient == null)
+        {
+            throw new ArgumentNullException(nameof(httpClient));
+        }
+
+        _config = configuration ?? throw new ArgumentNullException(nameof(configuration));
+        _client = new DefaultGraphQLHttpClient(httpClient);
+    }
+
+    public string SubgraphName => _config.SubgraphName;
+
+    public IAsyncEnumerable<GraphQLResponse> SubscribeAsync(
+        SubgraphGraphQLRequest request,
+        CancellationToken cancellationToken)
+    {
+        if (request is null)
+        {
+            throw new ArgumentNullException(nameof(request));
+        }
+
+        return SubscribeInternalAsync(request, cancellationToken);
+    }
+
+    private async IAsyncEnumerable<GraphQLResponse> SubscribeInternalAsync(
+        SubgraphGraphQLRequest subgraphRequest,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        var request = new GraphQLHttpRequest(subgraphRequest, _config.EndpointUri);
+        using var response = await _client.SendAsync(request, cancellationToken).ConfigureAwait(false);
+        
+        await foreach (var result in response.ReadAsResultStreamAsync(cancellationToken).ConfigureAwait(false))
+        {
+            yield return new GraphQLResponse(result);
+        }
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        _client.Dispose();
+        return default;
+    }
+}

src/HotChocolate/Fusion/src/Core/Clients/DefaultWebSocketGraphQLSubscriptionClientFactory.cs

@@ -6,23 +6,33 @@ namespace HotChocolate.Fusion.Clients;
 internal sealed class DefaultWebSocketGraphQLSubscriptionClientFactory
     : IGraphQLSubscriptionClientFactory
 {
+    private readonly IHttpClientFactory _httpClientFactory;
     private readonly IWebSocketConnectionFactory _connectionFactory;
 
     public DefaultWebSocketGraphQLSubscriptionClientFactory(
+        IHttpClientFactory httpClientFactory,
         IWebSocketConnectionFactory connectionFactory)
     {
+        _httpClientFactory = httpClientFactory ??
+            throw new ArgumentNullException(nameof(httpClientFactory));
         _connectionFactory = connectionFactory ??
             throw new ArgumentNullException(nameof(connectionFactory));
     }
 
     public IGraphQLSubscriptionClient CreateClient(IGraphQLClientConfiguration configuration)
     {
-        if (configuration is not WebSocketClientConfiguration webSocketClientConfig)
+        if (configuration is WebSocketClientConfiguration webSocketClientConfig)
         {
-            throw new ArgumentException(TransportConfigurationNotSupported, nameof(configuration));
+            var connection = _connectionFactory.CreateConnection(configuration.ClientName);
+            return new DefaultWebSocketGraphQLSubscriptionClient(webSocketClientConfig, connection);
         }
 
-        var connection = _connectionFactory.CreateConnection(configuration.ClientName);
-        return new DefaultWebSocketGraphQLSubscriptionClient(webSocketClientConfig, connection);
+        if (configuration is HttpClientConfiguration httpClientConfig)
+        {
+            var httpClient = _httpClientFactory.CreateClient(httpClientConfig.ClientName);
+            return new DefaultHttpGraphQLSubscriptionClient(httpClientConfig, httpClient);
+        }
+        
+        throw new ArgumentException(TransportConfigurationNotSupported, nameof(configuration));
     }
-}
+}

src/HotChocolate/Fusion/src/Core/DependencyInjection/FusionRequestExecutorBuilderExtensions.cs

@@ -47,6 +47,7 @@ public static FusionGatewayBuilder AddFusionGatewayServer(
                 sp.GetRequiredService<IHttpClientFactory>()));
         services.TryAddSingleton<IGraphQLSubscriptionClientFactory>(
             sp => new DefaultWebSocketGraphQLSubscriptionClientFactory(
+                sp.GetRequiredService<IHttpClientFactory>(),
                 sp.GetRequiredService<IWebSocketConnectionFactory>()));
 
         var builder = services
@@ -385,6 +386,14 @@ IGraphQLSubscriptionClient Create(IGraphQLClientConfiguration clientConfig)
             {
                 map2.Add(config.SubgraphName, () => Create(config));
             }
+            
+            foreach (var config in fusionGraphConfig.HttpClients)
+            {
+                if (!map2.ContainsKey(config.SubgraphName))
+                {
+                    map2.Add(config.SubgraphName, () => Create(config));
+                }
+            }
         }
 
         return new GraphQLClientFactory(map1, map2);

src/HotChocolate/Fusion/test/Core.Tests/DemoIntegrationTests.cs

@@ -1,11 +1,8 @@
 using CookieCrumble;
 using HotChocolate.Execution;
-using HotChocolate.Execution.Configuration;
 using HotChocolate.Fusion.Composition;
 using HotChocolate.Fusion.Composition.Features;
-using HotChocolate.Fusion.Planning;
 using HotChocolate.Fusion.Shared;
-using HotChocolate.Language;
 using HotChocolate.Skimmed.Serialization;
 using HotChocolate.Types.Relay;
 using Microsoft.Extensions.DependencyInjection;

src/HotChocolate/Fusion/test/Core.Tests/EventStreamTests.cs

@@ -0,0 +1,86 @@
+using CookieCrumble;
+using HotChocolate.Execution;
+using HotChocolate.Execution.Configuration;
+using HotChocolate.Fusion.Clients;
+using HotChocolate.Fusion.Composition;
+using HotChocolate.Fusion.Composition.Features;
+using HotChocolate.Fusion.Planning;
+using HotChocolate.Fusion.Shared;
+using HotChocolate.Language;
+using HotChocolate.Skimmed.Serialization;
+using HotChocolate.Types.Relay;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit.Abstractions;
+using static HotChocolate.Fusion.Shared.DemoProjectSchemaExtensions;
+using static HotChocolate.Language.Utf8GraphQLParser;
+using static HotChocolate.Fusion.TestHelper;
+
+namespace HotChocolate.Fusion;
+
+public class EventStreamTests
+{
+    private readonly Func<ICompositionLog> _logFactory;
+
+    public EventStreamTests(ITestOutputHelper output)
+    {
+        _logFactory = () => new TestCompositionLog(output);
+    }
+
+    [Fact]
+    public async Task Authors_And_Reviews_Subscription_OnNewReview()
+    {
+        // arrange
+        using var cts = new CancellationTokenSource(10_000);
+        using var demoProject = await DemoProject.CreateAsync(cts.Token);
+
+        // act
+        var fusionGraph = await new FusionGraphComposer(logFactory: _logFactory).ComposeAsync(
+            new[]
+            {
+                demoProject.Reviews2.ToConfiguration(Reviews2ExtensionSdl, onlyHttp: true),
+                demoProject.Accounts.ToConfiguration(AccountsExtensionSdl, onlyHttp: true)
+            },
+            default,
+            cts.Token);
+
+        var executor = await new ServiceCollection()
+            .AddSingleton(demoProject.HttpClientFactory)
+            .AddSingleton<IWebSocketConnectionFactory>(new NoWebSockets())
+            .AddFusionGatewayServer()
+            .ConfigureFromDocument(SchemaFormatter.FormatAsDocument(fusionGraph))
+            .BuildRequestExecutorAsync(cancellationToken: cts.Token);
+
+        var request = Parse(
+            """
+            subscription OnNewReview {
+                onNewReview {
+                    body
+                    author {
+                        name
+                    }
+                }
+            }
+            """);
+
+        // act
+        var result = await executor.ExecuteAsync(
+            QueryRequestBuilder
+                .New()
+                .SetQuery(request)
+                .Create(),
+            cts.Token);
+
+        // assert
+        var snapshot = new Snapshot();
+        await CollectStreamSnapshotData(snapshot, request, result, fusionGraph, cts.Token);
+        await snapshot.MatchAsync(cts.Token);
+    }
+    
+    private sealed class NoWebSockets : IWebSocketConnectionFactory
+    {
+        public IWebSocketConnection CreateConnection(string name)
+        {
+            throw new NotSupportedException();
+        }
+    }
+}