Publishing JSON format is broken; Mark as unsupported for now

Author: GermanCoding

src/Motor.Extensions.Hosting.PgMq/PgMqMessageProducer.cs

@@ -72,17 +72,11 @@ public async Task StartAsync(CancellationToken token = default)
     /// <param name="token">A cancellation token.</param>
     /// <returns>A <see cref="Task"/> that completes when the message has been added to the database.</returns>
     /// <remarks>
-    /// <list type="bullet">
-    /// <item>
-    /// In Protocol format (CloudEvents Binary Content Mode), CloudEvent attributes are written as Npgmq message
-    /// headers and the raw payload bytes are sent as the message body.
-    /// </item>
-    /// <item>
-    /// In JSON format (CloudEvents Structured Content Mode), the full CloudEvent – including all attributes and
-    /// the payload – is encoded as a single JSON envelope. No separate headers are used.
-    /// </item>
-    /// </list>
+    /// Only the protocol format (CloudEvents Binary Content Mode) is currently supported. CloudEvent attributes
+    /// are written as Npgmq message headers, and the raw payload bytes are sent as the message body.
+    /// The JSON mode is not yet supported by this producer.
     /// </remarks>
+    /// <exception cref="UnhandledCloudEventFormatException">If an unsupported cloud event format is specified.</exception>
     public async Task PublishMessageAsync(MotorCloudEvent<byte[]> motorCloudEvent, CancellationToken token = default)
     {
         switch (_publisherOptions.CloudEventFormat)
@@ -91,16 +85,6 @@ public async Task PublishMessageAsync(MotorCloudEvent<byte[]> motorCloudEvent, C
                 var headers = BuildHeaders(motorCloudEvent);
                 await _npgmqClient.SendAsync(_options.QueueName, motorCloudEvent.TypedData, headers, token);
                 break;
-            case CloudEventFormat.Json:
-                var jsonBytes = _cloudEventFormatter.EncodeStructuredModeMessage(
-                    motorCloudEvent.ConvertToCloudEvent(),
-                    out _
-                );
-                // Convert ReadOnlyMemory<byte> to string so that NpgmqClient stores the JSON as-is (JSONB).
-                // Passing the raw bytes directly would cause Npgmq to Base64-encode them via JsonSerializer.
-                var json = Encoding.UTF8.GetString(jsonBytes.ToArray());
-                await _npgmqClient.SendAsync(_options.QueueName, json, token);
-                break;
             default:
                 throw new UnhandledCloudEventFormatException(_publisherOptions.CloudEventFormat);
         }

test/Motor.Extensions.Hosting.PgMq_IntegrationTest/PgMqIntegrationTests.cs

@@ -49,7 +49,10 @@ public async Task Consume_ProduceAndConsumeProtocolFormat_ConsumedDataEqualsPubl
         await producer.PublishMessageAsync(
             MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes(expectedMessage))
         );
-        var received = await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task ? await tcs.Task : null;
+        var received =
+            await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task
+                ? await tcs.Task
+                : null;
         await cts.CancelAsync();
         await executeTask;
         await consumer.StopAsync();
@@ -58,31 +61,18 @@ await producer.PublishMessageAsync(
     }
 
     [Fact(Timeout = 50000)]
-    public async Task Consume_ProduceAndConsumeJsonFormat_ConsumedDataEqualsPublished()
+    public async Task Consume_ProduceAndConsumeJsonFormat_ThrowsUnhandledCloudEventFormatException()
     {
         const string expectedMessage = "hello-json";
         var queueName = _randomizerString.Generate()!;
         await using var producer = GetProducer<string>(queueName, CloudEventFormat.Json);
-        await using var consumer = GetConsumer<string>(queueName);
         await producer.StartAsync();
-        await consumer.StartAsync();
-        var tcs = new TaskCompletionSource<byte[]>();
-        consumer.ConsumeCallbackAsync = (evt, _) =>
+        await Assert.ThrowsAsync<UnhandledCloudEventFormatException>(async () =>
         {
-            tcs.TrySetResult(evt.TypedData);
-            return Task.FromResult(ProcessedMessageStatus.Success);
-        };
-        using var cts = new CancellationTokenSource();
-        var executeTask = consumer.ExecuteAsync(cts.Token);
-        await producer.PublishMessageAsync(
-            MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes(expectedMessage))
-        );
-        var received = await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task ? await tcs.Task : null;
-        await cts.CancelAsync();
-        await executeTask;
-        await consumer.StopAsync();
-        Assert.NotNull(received);
-        Assert.Equal(expectedMessage, Encoding.UTF8.GetString(received!));
+            await producer.PublishMessageAsync(
+                MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes(expectedMessage))
+            );
+        });
     }
 
     [Fact(Timeout = 50000)]
@@ -269,7 +259,10 @@ public async Task Consume_SuccessfulMessage_ProtocolFormat_CloudEventAttributesR
             null
         );
         await producer.PublishMessageAsync(cloudEvent);
-        var receivedEvent = await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task ? await tcs.Task : null;
+        var receivedEvent =
+            await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task
+                ? await tcs.Task
+                : null;
         await cts.CancelAsync();
         await executeTask;
         await consumer.StopAsync();