Rewrite OutgoingRabbitMQChannel to use SenderProcessor for graceful shutdown

- Fix RabbitMQMessageConverter to serialize complex objects as JSON
- Re-enable disabled tests, replace Thread.sleep with Awaitility

Author: ozangunalp

smallrye-reactive-messaging-rabbitmq-og/pom.xml

@@ -92,6 +92,13 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- Jackson Databind for JSON serialization of complex payloads -->
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>io.smallrye.reactive</groupId>
       <artifactId>smallrye-reactive-messaging-in-memory</artifactId>

smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/RabbitMQConnector.java

@@ -203,38 +203,7 @@ public Flow.Subscriber<? extends Message<?>> getSubscriber(Config config) {
 
         outgoings.add(channel);
 
-        // Wrap Reactive Streams Subscriber as Flow.Subscriber
-        return new Flow.Subscriber<Message<?>>() {
-            @Override
-            public void onSubscribe(Flow.Subscription subscription) {
-                channel.onSubscribe(new org.reactivestreams.Subscription() {
-                    @Override
-                    public void request(long n) {
-                        subscription.request(n);
-                    }
-
-                    @Override
-                    public void cancel() {
-                        subscription.cancel();
-                    }
-                });
-            }
-
-            @Override
-            public void onNext(Message<?> message) {
-                channel.onNext(message);
-            }
-
-            @Override
-            public void onError(Throwable throwable) {
-                channel.onError(throwable);
-            }
-
-            @Override
-            public void onComplete() {
-                channel.onComplete();
-            }
-        };
+        return channel.getSink();
     }
 
     @Override
@@ -286,12 +255,7 @@ public void terminate(
 
         // Clean up all outgoing channels
         for (io.smallrye.reactive.messaging.rabbitmq.og.internals.OutgoingRabbitMQChannel outgoing : outgoings) {
-            try {
-                outgoing.onComplete();
-            } catch (Exception e) {
-                // Log but continue cleanup
-                e.printStackTrace();
-            }
+            outgoing.closeQuietly();
         }
         outgoings.clear();
 

smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/RabbitMQMessageConverter.java

@@ -10,6 +10,8 @@
 
 import com.rabbitmq.client.AMQP;
 
+import io.vertx.core.json.JsonObject;
+
 /**
  * Utility class for converting between Reactive Messaging Message and RabbitMQ message format.
  * Handles different payload types and content-type determination.
@@ -166,9 +168,8 @@ private static byte[] getBodyFromPayload(Object payload) {
             return payload.toString().getBytes(StandardCharsets.UTF_8);
         }
 
-        // For other types, convert to string representation
-        // In a full implementation, you might want to use Jackson for JSON serialization
-        return payload.toString().getBytes(StandardCharsets.UTF_8);
+        // For complex objects, serialize to JSON
+        return JsonObject.mapFrom(payload).encode().getBytes(StandardCharsets.UTF_8);
     }
 
     /**