Add Federated Event Streams

dictionary.txt

@@ -229,8 +229,10 @@ RRGGBB
 Ruckstuhl
 runbooks
 sandboxed
+SASL
 Satisfiability
 Senn
+servicebus
 shoooe
 SHORTSHA
 Silmarillion

src/All.slnx

@@ -221,6 +221,10 @@
     <Project Path="HotChocolate/Fusion/src/Fusion.Execution/HotChocolate.Fusion.Execution.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Language/HotChocolate.Fusion.Language.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Packaging/HotChocolate.Fusion.Packaging.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.AmazonSqs/HotChocolate.Fusion.Subscriptions.AmazonSqs.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.AzureEventHubs/HotChocolate.Fusion.Subscriptions.AzureEventHubs.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.Kafka/HotChocolate.Fusion.Subscriptions.Kafka.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.NATS/HotChocolate.Fusion.Subscriptions.NATS.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Utilities/HotChocolate.Fusion.Utilities.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.SourceSchema.Packaging/HotChocolate.Fusion.SourceSchema.Packaging.csproj" />
   </Folder>
@@ -231,6 +235,10 @@
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.ApolloFederation.Compliance.Tests/HotChocolate.Fusion.Connectors.ApolloFederation.Compliance.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.ApolloFederation.Tests/HotChocolate.Fusion.Connectors.ApolloFederation.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.InMemory.Tests/HotChocolate.Fusion.Connectors.InMemory.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.AmazonSqs.Tests/HotChocolate.Fusion.Subscriptions.AmazonSqs.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.AzureEventHubs.Tests/HotChocolate.Fusion.Subscriptions.AzureEventHubs.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.Kafka.Tests/HotChocolate.Fusion.Subscriptions.Kafka.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.NATS.Tests/HotChocolate.Fusion.Subscriptions.NATS.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Caching.Tests/HotChocolate.Fusion.Caching.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Diagnostics.Tests/HotChocolate.Fusion.Diagnostics.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Execution.Tests/HotChocolate.Fusion.Execution.Tests.csproj" />

src/Directory.Packages.props

@@ -6,6 +6,10 @@
     <PackageVersion Include="Aspire.Hosting.AppHost" Version="13.1.2" />
     <PackageVersion Include="Aspire.Hosting.PostgreSQL" Version="13.1.2" />
     <PackageVersion Include="Aspire.Hosting" Version="13.1.2" />
+    <PackageVersion Include="AWSSDK.SimpleNotificationService" Version="4.0.3.6" />
+    <PackageVersion Include="AWSSDK.SQS" Version="4.0.3.5" />
+    <PackageVersion Include="Azure.Identity" Version="1.13.2" />
+    <PackageVersion Include="Azure.Messaging.EventHubs" Version="5.12.2" />
     <PackageVersion Include="Azure.Storage.Blobs" Version="12.23.0" />
     <PackageVersion Include="Basic.Reference.Assemblies.Net110" Version="1.8.8" />
     <PackageVersion Include="Basic.Reference.Assemblies.Net100" Version="1.8.8" />
@@ -15,6 +19,7 @@
     <PackageVersion Include="BenchmarkDotNet.Diagnostics.dotMemory" Version="0.15.8" />
     <PackageVersion Include="ChilliCream.ModelContextProtocol.AspNetCore" Version="1.4.0" />
     <PackageVersion Include="ChilliCream.Nitro.App" Version="$(NitroVersion)" />
+    <PackageVersion Include="Confluent.Kafka" Version="2.14.2" />
     <PackageVersion Include="DiffPlex" Version="1.9.0" />
     <PackageVersion Include="Duende.IdentityModel" Version="7.1.0" />
     <PackageVersion Include="Duende.IdentityModel.OidcClient" Version="6.0.1" />
@@ -23,6 +28,7 @@
     <PackageVersion Include="JsonSchema.Net" Version="7.3.4" />
     <PackageVersion Include="Marten" Version="8.37.0" />
     <PackageVersion Include="McMaster.Extensions.CommandLineUtils" Version="4.0.1" />
+    <PackageVersion Include="MessagePack" Version="2.5.301" />
     <PackageVersion Include="Microsoft.Azure.Functions.Extensions" Version="1.1.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Core" Version="2.0.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Abstractions" Version="1.1.0" />
@@ -39,6 +45,7 @@
     <PackageVersion Include="MongoDB.Driver" Version="3.0.0" />
     <PackageVersion Include="Moq" Version="4.20.72" />
     <PackageVersion Include="NATS.Client.Core" Version="2.7.2" />
+    <PackageVersion Include="NATS.Client.JetStream" Version="2.7.2" />
     <PackageVersion Include="NATS.Extensions.Microsoft.DependencyInjection" Version="2.7.2" />
     <PackageVersion Include="NetTopologySuite" Version="2.0.0" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.2" />
@@ -74,6 +81,7 @@
     <PackageVersion Include="System.Net.ServerSentEvents" Version="10.0.0" />
     <PackageVersion Include="System.Reactive" Version="6.1.0" />
     <PackageVersion Include="Testcontainers" Version="4.11.0" />
+    <PackageVersion Include="Testcontainers.LocalStack" Version="4.12.0" />
     <PackageVersion Include="xunit.assert" Version="2.9.3" />
     <PackageVersion Include="xunit.extensibility.core" Version="2.9.3" />
     <PackageVersion Include="xunit.extensibility.execution" Version="2.9.3" />

src/HotChocolate/Core/src/Types/Types/Composite/Directives/EventCursorAttribute.cs

@@ -0,0 +1,48 @@
+using System.Reflection;
+using HotChocolate.Types.Descriptors;
+
+namespace HotChocolate.Types.Composite;
+
+/// <summary>
+/// <para>
+/// Applies the @eventCursor directive. On an argument it marks the resume input that
+/// the distributed GraphQL executor uses to continue an event stream. On an output
+/// field it marks the cursor that carries the position within the stream.
+/// </para>
+/// <para>
+/// @eventCursor
+/// </para>
+/// </summary>
+[AttributeUsage(
+    AttributeTargets.Parameter
+    | AttributeTargets.Property,
+    AllowMultiple = false)]
+public sealed class EventCursorAttribute : DescriptorAttribute
+{
+    protected internal override void TryConfigure(
+        IDescriptorContext context,
+        IDescriptor descriptor,
+        ICustomAttributeProvider? attributeProvider)
+    {
+        switch (descriptor)
+        {
+            case IArgumentDescriptor arg:
+                arg.EventCursor();
+                break;
+
+            case IObjectFieldDescriptor field:
+                field.EventCursor();
+                break;
+
+            default:
+                throw new SchemaException(
+                    SchemaErrorBuilder.New()
+                        .SetMessage(
+                            "EventCursor directive is only supported on arguments and "
+                            + "field definitions of object types.")
+                        .SetExtension("member", attributeProvider)
+                        .SetExtension("descriptor", descriptor)
+                        .Build());
+        }
+    }
+}

src/HotChocolate/Core/src/Types/Types/Composite/Directives/EventCursorDirective.cs

@@ -0,0 +1,33 @@
+namespace HotChocolate.Types.Composite;
+
+/// <summary>
+/// <para>
+/// The @eventCursor directive marks the cursor of an event stream. On a subscription
+/// field argument it marks the resume input that the distributed executor uses to
+/// continue a stream after a previously received event. On an output field it marks
+/// the value that carries the cursor of each emitted event, which a client can store
+/// and later pass back to resume the stream.
+/// </para>
+/// <para>
+/// directive @eventCursor on ARGUMENT_DEFINITION | FIELD_DEFINITION
+/// </para>
+/// </summary>
+[DirectiveType(
+    DirectiveNames.EventCursor.Name,
+    DirectiveLocation.ArgumentDefinition
+    | DirectiveLocation.FieldDefinition,
+    IsRepeatable = false)]
+public sealed class EventCursorDirective
+{
+    private EventCursorDirective()
+    {
+    }
+
+    /// <summary>
+    /// The singleton instance of the <see cref="EventCursorDirective"/> directive.
+    /// </summary>
+    public static EventCursorDirective Instance { get; } = new();
+
+    /// <inheritdoc />
+    public override string ToString() => "@eventCursor";
+}

src/HotChocolate/Core/src/Types/Types/Composite/Directives/EventStream.cs

@@ -0,0 +1,26 @@
+namespace HotChocolate.Types.Composite;
+
+/// <summary>
+/// Provides a placeholder for the body of an event-stream subscription field. Event-stream
+/// fields are fulfilled by the distributed GraphQL executor and never execute their local
+/// resolver, so this helper only exists to let the authored field compile.
+/// </summary>
+public static class EventStream
+{
+    /// <summary>
+    /// Acts as a placeholder body for an event-stream subscription field.
+    /// </summary>
+    /// <typeparam name="T">The event payload type.</typeparam>
+    /// <param name="args">
+    /// The arguments for the event-stream subscription field.
+    /// </param>
+    /// <returns>This method never returns a value.</returns>
+    /// <exception cref="NotSupportedException">
+    /// Always thrown. Event-stream fields are fulfilled by the distributed GraphQL executor,
+    /// not the local resolver.
+    /// </exception>
+    public static T Create<T>(params object?[] args)
+        => throw new NotSupportedException(
+            "Event-stream fields are fulfilled by the Fusion gateway broker, "
+            + "not the local resolver.");
+}

src/HotChocolate/Core/src/Types/Types/Composite/Directives/EventStreamAttribute.cs

@@ -0,0 +1,58 @@
+using System.Reflection;
+using HotChocolate.Types.Descriptors;
+
+namespace HotChocolate.Types.Composite;
+
+/// <summary>
+/// <para>
+/// Applies the @eventStream directive to this subscription field to declare that it is
+/// fulfilled by an event stream behind the distributed GraphQL executor.
+/// </para>
+/// <para>
+/// @eventStream(message: "user { id }", topics: ["onUserCreated"], broker: "kafka")
+/// </para>
+/// </summary>
+[AttributeUsage(AttributeTargets.Method)]
+public class EventStreamAttribute : ObjectFieldDescriptorAttribute
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="EventStreamAttribute"/> class.
+    /// </summary>
+    /// <param name="message">
+    /// The payload selection set. The outer braces are optional and will be added if not present.
+    /// </param>
+    public EventStreamAttribute(string message)
+    {
+        Message = message;
+    }
+
+    /// <summary>
+    /// Gets the payload selection set.
+    /// </summary>
+    public string Message { get; }
+
+    /// <summary>
+    /// Gets or sets the topics the event stream subscribes to.
+    /// </summary>
+    public string[]? Topics { get; set; }
+
+    /// <summary>
+    /// Gets or sets a single topic the event stream subscribes to.
+    /// </summary>
+    public string? Topic
+    {
+        get => Topics is { Length: > 0 } ? Topics[0] : null;
+        set => Topics = value is null ? null : [value];
+    }
+
+    /// <summary>
+    /// Gets or sets the broker that provides the event stream.
+    /// </summary>
+    public string? Broker { get; set; }
+
+    protected override void OnConfigure(
+        IDescriptorContext context,
+        IObjectFieldDescriptor descriptor,
+        MemberInfo? member)
+        => descriptor.EventStream(Message, Topics, Broker);
+}

src/HotChocolate/Core/src/Types/Types/Composite/Directives/EventStreamDescriptorExtensions.cs

@@ -0,0 +1,128 @@
+using HotChocolate.Language;
+
+namespace HotChocolate.Types.Composite;
+
+/// <summary>
+/// Provides extension methods to configure the @eventStream and @eventCursor directives
+/// with the fluent API.
+/// </summary>
+public static class EventStreamDescriptorExtensions
+{
+    /// <summary>
+    /// <para>
+    /// Applies the @eventStream directive to this subscription field to declare that it is
+    /// fulfilled by an event stream behind the distributed GraphQL executor.
+    /// </para>
+    /// <para>
+    /// @eventStream(message: "user { id }", topics: ["onUserCreated"], broker: "kafka")
+    /// </para>
+    /// </summary>
+    /// <param name="descriptor">The object field descriptor.</param>
+    /// <param name="message">
+    /// The payload selection set. The outer braces are optional and will be added if not present.
+    /// </param>
+    /// <param name="topic">The topic the event stream subscribes to.</param>
+    /// <param name="broker">The broker that provides the event stream.</param>
+    /// <returns>The object field descriptor with the @eventStream directive applied.</returns>
+    public static IObjectFieldDescriptor EventStream(
+        this IObjectFieldDescriptor descriptor,
+        string message,
+        string? topic = null,
+        string? broker = null)
+    {
+        ArgumentNullException.ThrowIfNull(descriptor);
+        ArgumentNullException.ThrowIfNull(message);
+
+        var topics = topic is null ? null : new[] { topic };
+
+        return ApplyEventStream(descriptor, message, topics, broker);
+    }
+
+    /// <summary>
+    /// <para>
+    /// Applies the @eventStream directive to this subscription field to declare that it is
+    /// fulfilled by an event stream behind the distributed GraphQL executor.
+    /// </para>
+    /// <para>
+    /// @eventStream(message: "user { id }", topics: ["onUserCreated"], broker: "kafka")
+    /// </para>
+    /// </summary>
+    /// <param name="descriptor">The object field descriptor.</param>
+    /// <param name="message">
+    /// The payload selection set. The outer braces are optional and will be added if not present.
+    /// </param>
+    /// <param name="topics">The topics the event stream subscribes to.</param>
+    /// <param name="broker">The broker that provides the event stream.</param>
+    /// <returns>The object field descriptor with the @eventStream directive applied.</returns>
+    public static IObjectFieldDescriptor EventStream(
+        this IObjectFieldDescriptor descriptor,
+        string message,
+        string[]? topics,
+        string? broker = null)
+    {
+        ArgumentNullException.ThrowIfNull(descriptor);
+        ArgumentNullException.ThrowIfNull(message);
+
+        return ApplyEventStream(descriptor, message, topics, broker);
+    }
+
+    private static IObjectFieldDescriptor ApplyEventStream(
+        IObjectFieldDescriptor descriptor,
+        string message,
+        IReadOnlyList<string>? topics,
+        string? broker)
+    {
+        SelectionSetNode selectionSet;
+
+        try
+        {
+            selectionSet = FieldSelectionSetType.ParseSelectionSet(message);
+        }
+        catch (SyntaxException ex)
+        {
+            descriptor.Extend().OnBeforeNaming(
+                (ctx, _) => ctx.ReportError(
+                    SchemaErrorBuilder.New()
+                        .SetMessage("The field selection set syntax is invalid.")
+                        .SetException(ex)
+                        .Build()));
+            return descriptor;
+        }
+
+        return descriptor.Directive(new EventStreamDirective(selectionSet, topics, broker));
+    }
+
+    /// <summary>
+    /// <para>
+    /// Applies the @eventCursor directive to this argument to mark it as the resume input
+    /// that the distributed GraphQL executor uses to continue an event stream.
+    /// </para>
+    /// <para>
+    /// @eventCursor
+    /// </para>
+    /// </summary>
+    /// <param name="descriptor">The argument descriptor.</param>
+    /// <returns>The argument descriptor with the @eventCursor directive applied.</returns>
+    public static IArgumentDescriptor EventCursor(this IArgumentDescriptor descriptor)
+    {
+        ArgumentNullException.ThrowIfNull(descriptor);
+        return descriptor.Directive(EventCursorDirective.Instance);
+    }
+
+    /// <summary>
+    /// <para>
+    /// Applies the @eventCursor directive to this output field to mark it as the cursor
+    /// that carries the position within an event stream.
+    /// </para>
+    /// <para>
+    /// @eventCursor
+    /// </para>
+    /// </summary>
+    /// <param name="descriptor">The object field descriptor.</param>
+    /// <returns>The object field descriptor with the @eventCursor directive applied.</returns>
+    public static IObjectFieldDescriptor EventCursor(this IObjectFieldDescriptor descriptor)
+    {
+        ArgumentNullException.ThrowIfNull(descriptor);
+        return descriptor.Directive(EventCursorDirective.Instance);
+    }
+}