Added better error when expected message type does not match actual m…

…essage type. (#6086)

src/HotChocolate/Core/src/Subscriptions.InMemory/InMemoryPubSub.cs

@@ -24,15 +24,15 @@ protected override async ValueTask OnSendAsync<TMessage>(
         TMessage message,
         CancellationToken cancellationToken = default)
     {
-        if (TryGetTopic<InMemoryTopic<TMessage>>(formattedTopic, out var topic))
+        if (TryGetTopic<TMessage>(formattedTopic, out var topic))
         {
             await topic.PublishAsync(message, cancellationToken).ConfigureAwait(false);
         }
     }
 
     protected override ValueTask OnCompleteAsync(string formattedTopic)
     {
-        if (TryGetTopic<ITopic>(formattedTopic, out var topic))
+        if (TryGetTopic(formattedTopic, out var topic))
         {
             topic.TryComplete();
         }

src/HotChocolate/Core/src/Subscriptions/DefaultPubSub.cs

@@ -12,7 +12,7 @@ namespace HotChocolate.Subscriptions;
 public abstract class DefaultPubSub : ITopicEventReceiver, ITopicEventSender, IDisposable
 {
     private readonly SemaphoreSlim _subscribeSemaphore = new(1, 1);
-    private readonly ConcurrentDictionary<string, IDisposable> _topics = new(Ordinal);
+    private readonly ConcurrentDictionary<string, ITopic> _topics = new(Ordinal);
     private readonly TopicFormatter _topicFormatter;
     private readonly ISubscriptionDiagnosticEvents _diagnosticEvents;
     private bool _disposed;
@@ -119,7 +119,7 @@ await CreateTopicAsync<TMessage>(
         return sourceStream;
 
         static async ValueTask<ISourceStream<TMessage>?> TryCreateSourceStream(
-            IDisposable topic,
+            ITopic topic,
             CancellationToken cancellationToken)
         {
             if (topic is DefaultTopic<TMessage> et)
@@ -129,7 +129,7 @@ await CreateTopicAsync<TMessage>(
 
             // we found a topic with the same name but a different message type.
             // this is an invalid state and we will except.
-            throw new InvalidMessageTypeException();
+            throw new InvalidMessageTypeException(topic.MessageType, typeof(TMessage));
         }
     }
 
@@ -195,19 +195,33 @@ protected abstract DefaultTopic<TMessage> OnCreateTopic<TMessage>(
     protected virtual string FormatTopicName(string topic)
         => _topicFormatter.Format(topic);
 
-    protected bool TryGetTopic<TTopic>(
+    protected bool TryGetTopic<TMessage>(
         string formattedTopic,
-        [NotNullWhen(true)] out TTopic? topic)
+        [NotNullWhen(true)] out DefaultTopic<TMessage>? topic)
     {
         if (_topics.TryGetValue(formattedTopic, out var value))
         {
-            if (value is TTopic casted)
+            if (value is DefaultTopic<TMessage> casted)
             {
                 topic = casted;
                 return true;
             }
 
-            throw new InvalidMessageTypeException();
+            throw new InvalidMessageTypeException(value.MessageType, typeof(TMessage));
+        }
+
+        topic = default;
+        return false;
+    }
+
+    protected bool TryGetTopic(
+        string formattedTopic,
+        [NotNullWhen(true)] out ITopic? topic)
+    {
+        if (_topics.TryGetValue(formattedTopic, out var value))
+        {
+            topic = value;
+            return true;
         }
 
         topic = default;

src/HotChocolate/Core/src/Subscriptions/DefaultTopic.cs

@@ -67,10 +67,15 @@ protected DefaultTopic(
     }
 
     /// <summary>
-    /// The name of this topic.
+    /// Gets the name of this topic.
     /// </summary>
     public string Name { get; }
 
+    /// <summary>
+    /// Gets the message type of this topic.
+    /// </summary>
+    public Type MessageType => typeof(TMessage);
+
     /// <summary>
     /// Allows access to the diagnostic events.
     /// </summary>
@@ -461,8 +466,3 @@ public void Dispose() { }
         public static readonly DefaultSession Instance = new();
     }
 }
-
-public interface ITopic
-{
-    void TryComplete();
-}

src/HotChocolate/Core/src/Subscriptions/ITopic.cs

@@ -0,0 +1,17 @@
+namespace HotChocolate.Subscriptions;
+
+/// <summary>
+/// Represents a topic.
+/// </summary>
+public interface ITopic : IDisposable
+{
+    /// <summary>
+    /// Gets the message type of this topic.
+    /// </summary>
+    public Type MessageType { get; }
+
+    /// <summary>
+    /// Allows to complete a topic.
+    /// </summary>
+    void TryComplete();
+}

src/HotChocolate/Core/src/Subscriptions/InvalidMessageTypeException.cs

@@ -1,4 +1,4 @@
-using System.Runtime.Serialization;
+using HotChocolate.Subscriptions.Properties;
 
 namespace HotChocolate.Subscriptions;
 
@@ -9,16 +9,37 @@ namespace HotChocolate.Subscriptions;
 [Serializable]
 public class InvalidMessageTypeException : Exception
 {
-    public InvalidMessageTypeException() { }
+    /// <summary>
+    /// Initializes a new instance of the <see cref="InvalidMessageTypeException"/> class.
+    /// </summary>
+    /// <param name="topicMessageType">
+    /// The topic message type.
+    /// </param>
+    /// <param name="requestedMessageType">
+    /// The requested message type.
+    /// </param>
+    public InvalidMessageTypeException(Type topicMessageType, Type requestedMessageType)
+        : base(CreateMessage(topicMessageType, requestedMessageType))
+    {
+        TopicMessageType = topicMessageType;
+        RequestedMessageType = requestedMessageType;
+    }
 
-    public InvalidMessageTypeException(string message)
-        : base(message) { }
+    /// <summary>
+    /// Gets the topic message type.
+    /// </summary>
+    public Type TopicMessageType { get; }
 
-    public InvalidMessageTypeException(string message, Exception inner)
-        : base(message, inner) { }
+    /// <summary>
+    /// Gets the requested message type.
+    /// </summary>
+    public Type RequestedMessageType { get; }
 
-    protected InvalidMessageTypeException(
-        SerializationInfo info,
-        StreamingContext context)
-        : base(info, context) { }
+    private static string CreateMessage(
+        Type topicMessageType,
+        Type requestedMessageType)
+        => string.Format(
+            Resources.InvalidMessageTypeException_Message,
+            topicMessageType.FullName,
+            requestedMessageType.FullName);
 }

src/HotChocolate/Core/src/Subscriptions/Properties/Resources.resx

@@ -113,4 +113,7 @@
   <data name="MessageEnvelope_UnsubscribeAndComplete_DoNotHaveBody" xml:space="preserve">
     <value>Complete and Unsubscribe messages do not have a body.</value>
   </data>
+  <data name="InvalidMessageTypeException_Message" xml:space="preserve">
+    <value>The topic already exists with a different message type. Topic message type: {0}. Requested message type: {1}.</value>
+  </data>
 </root>

src/HotChocolate/Core/test/Subscriptions.InMemory.Tests/InMemoryIntegrationTests.cs

@@ -1,4 +1,6 @@
-using System.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
+using HotChocolate.Execution;
 using Microsoft.Extensions.DependencyInjection;
 using HotChocolate.Execution.Configuration;
 using Xunit.Abstractions;
@@ -36,6 +38,32 @@ public override Task Subscribe_Topic_With_Arguments_2_Topics()
     public override Task Subscribe_Topic_With_2_Arguments()
         => base.Subscribe_Topic_With_2_Arguments();
 
+    [Fact]
+    public virtual async Task Invalid_Message_Type()
+    {
+        // arrange
+        using var cts = new CancellationTokenSource(5000);
+        await using var services = CreateServer<Subscription3>();
+        var sender = services.GetRequiredService<ITopicEventSender>();
+
+        var result = await services.ExecuteRequestAsync(
+                "subscription { onMessage2(arg1: \"a\", arg2: \"b\") }",
+                cancellationToken: cts.Token)
+            .ConfigureAwait(false);
+
+        // we need to execute the read for the subscription to start receiving.
+        await using var responseStream = result.ExpectResponseStream();
+        var results = responseStream.ReadResultsAsync().ConfigureAwait(false);
+
+        // act
+        async Task Send() => await sender.SendAsync("OnMessage2_a_b", 1, cts.Token).ConfigureAwait(false);
+
+        // assert
+        var exception = await Assert.ThrowsAsync<InvalidMessageTypeException>(Send);
+        Assert.Equal(typeof(string), exception.TopicMessageType);
+        Assert.Equal(typeof(int), exception.RequestedMessageType);
+    }
+
     protected override void ConfigurePubSub(IRequestExecutorBuilder graphqlBuilder)
         => graphqlBuilder.AddInMemorySubscriptions();
 }