Merge pull request #40 from SolaceLabs/issues

Fixed #39 #36

pubsub-plus-connector/src/main/java/com/solace/quarkus/messaging/incoming/SolaceInboundMessage.java

@@ -4,6 +4,8 @@
 
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
 
 import org.eclipse.microprofile.reactive.messaging.Metadata;
 
@@ -82,6 +84,16 @@ public Metadata getMetadata() {
         return metadata;
     }
 
+    @Override
+    public Supplier<CompletionStage<Void>> getAck() {
+        return this::ack;
+    }
+
+    @Override
+    public Function<Throwable, CompletionStage<Void>> getNack() {
+        return this::nack;
+    }
+
     @Override
     public CompletionStage<Void> ack() {
         this.unacknowledgedMessageTracker.decrement();

pubsub-plus-connector/src/test/java/com/solace/quarkus/messaging/locals/LocalPropagationTest.java

@@ -7,7 +7,6 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
@@ -196,14 +195,7 @@ public Message<Integer> process(Message<String> input) {
             Vertx.currentContext().putLocal("uuid", value);
             Vertx.currentContext().putLocal("input", payload);
 
-            return input.withPayload(payload + 1)
-                    .withAck(() -> {
-                        CompletableFuture<Void> cf = new CompletableFuture<>();
-                        executor.execute(() -> {
-                            cf.complete(null);
-                        });
-                        return cf;
-                    });
+            return input.withPayload(payload + 1);
         }
 
         @Incoming("process")

runtime/src/main/resources/META-INF/quarkus-extension.yaml

@@ -1,10 +1,12 @@
 name: Solace
 description: Solace Quarkus Extension for integrating with Solace PubSub+ message brokers. The extension provides the ability to publish or consume events from event mesh.
 metadata:
-#  keywords:
-#    - solace
-#    - pubsubplus event broker
+  keywords:
+    - solace
+    - pubsubplus event broker
+    - event driven integration
+    - reactive-solace
 #  guide: https://quarkiverse.github.io/quarkiverse-docs/solace/dev/ # To create and publish this guide, see https://github.com/quarkiverse/quarkiverse/wiki#documenting-your-extension
-#  categories:
-#    - "miscellaneous"
-#  status: "preview"
+  categories:
+    - "messaging"
+  status: "preview"

samples/hello-connector-solace/src/main/java/com/solace/quarkus/samples/HelloConsumer.java

@@ -1,7 +1,7 @@
 package com.solace.quarkus.samples;
 
 import java.nio.charset.StandardCharsets;
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 import jakarta.enterprise.context.ApplicationScoped;
 
@@ -11,62 +11,55 @@
 import com.solace.quarkus.messaging.outgoing.SolaceOutboundMetadata;
 
 import io.quarkus.logging.Log;
+import io.smallrye.mutiny.Multi;
 
 @ApplicationScoped
 public class HelloConsumer {
+
     /**
-     * Publish a simple message using TryMe in Solace broker and you should see the message published to topic
+     * Publishes message to topic hello/foobar which is subscribed by queue.foobar
+     *
+     * @see #consumeMessage(SolaceInboundMessage)
+     * @return
+     */
+    @Outgoing("hello-out")
+    Multi<Message<String>> publishMessage() {
+        SolaceOutboundMetadata outboundMetadata = SolaceOutboundMetadata.builder()
+                .setApplicationMessageId("1").createPubSubOutboundMetadata();
+        return Multi.createFrom().items("1").map(m -> Message.of(m, Metadata.of(outboundMetadata)));
+    }
+
+    /**
+     * Receives message from queue - queue.foobar
      *
      * @param p
      */
     @Incoming("hello-in")
-    @Outgoing("hello-out")
     @Acknowledgment(Acknowledgment.Strategy.MANUAL)
-    Message<?> consumeAndPublish(SolaceInboundMessage<?> p) {
-        Log.infof("Received message: %s", new String(p.getMessage().getPayloadAsBytes(), StandardCharsets.UTF_8));
-        SolaceOutboundMetadata outboundMetadata = SolaceOutboundMetadata.builder()
-                .createPubSubOutboundMetadata();
-        Message<?> outboundMessage = Message.of(p.getPayload(), Metadata.of(outboundMetadata), () -> {
-            CompletableFuture completableFuture = new CompletableFuture();
-            p.ack();
-            completableFuture.complete(null);
-            return completableFuture;
-        }, (throwable) -> {
-            CompletableFuture completableFuture = new CompletableFuture();
-            p.nack(throwable, p.getMetadata());
-            completableFuture.complete(null);
-            return completableFuture;
-        });
-        return outboundMessage;
+    CompletionStage<Void> consumeMessage(SolaceInboundMessage<?> p) {
+        Log.infof("Received message: %s from topic: %s", new String(p.getMessage().getPayloadAsBytes(), StandardCharsets.UTF_8),
+                p.getMessage().getDestinationName());
+        return p.ack();
     }
 
     /**
-     * Publish a simple string from using TryMe in Solace broker and you should see the message published to dynamic destination
-     * topic
+     * Receives message from queue - queue.dynamic.topic and overwrites the topic configured in outgoing channel
+     * dynamic-destination-out
+     *
+     * See [resources/application.properties#mp.messaging.outgoing.dynamic-destination-out.producer.topic]
      *
      * @param p
      */
     @Incoming("dynamic-destination-in")
     @Outgoing("dynamic-destination-out")
-    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
     Message<?> consumeAndPublishToDynamicTopic(SolaceInboundMessage<?> p) {
-        Log.infof("Received message: %s", new String(p.getMessage().getPayloadAsBytes(), StandardCharsets.UTF_8));
+        Log.infof("Received message: %s from topic: %s", new String(p.getMessage().getPayloadAsBytes(), StandardCharsets.UTF_8),
+                p.getMessage().getDestinationName());
         SolaceOutboundMetadata outboundMetadata = SolaceOutboundMetadata.builder()
                 .setApplicationMessageId("test")
-                .setDynamicDestination("solace/quarkus/producer/" + p.getMessage().getCorrelationId()) // make sure correlationID is available on incoming message
+                .setDynamicDestination("hello/foobar/" + p.getMessage().getApplicationMessageId())
                 .createPubSubOutboundMetadata();
-        Message<?> outboundMessage = Message.of(p.getPayload(), Metadata.of(outboundMetadata), () -> {
-            CompletableFuture completableFuture = new CompletableFuture();
-            p.ack();
-            completableFuture.complete(null);
-            return completableFuture;
-        }, (throwable) -> {
-            CompletableFuture completableFuture = new CompletableFuture();
-            p.nack(throwable, p.getMetadata());
-            completableFuture.complete(null);
-            return completableFuture;
-        });
-        return outboundMessage;
+        return p.addMetadata(outboundMetadata);
     }
 
 }

samples/hello-connector-solace/src/main/java/com/solace/quarkus/samples/PublisherResource.java

@@ -15,7 +15,7 @@
 @Path("/hello")
 public class PublisherResource {
 
-    @Channel("hello")
+    @Channel("hello-out")
     MutinyEmitter<Person> foobar;
 
     /**

samples/hello-connector-solace/src/main/resources/application.properties

@@ -1,24 +1,27 @@
-quarkus.solace.host=
-quarkus.solace.vpn=
-quarkus.solace.authentication.basic.username=
-quarkus.solace.authentication.basic.password=
+#quarkus.solace.host=
+#quarkus.solace.vpn=
+#quarkus.solace.authentication.basic.username=
+#quarkus.solace.authentication.basic.password=
 
 mp.messaging.outgoing.hello-out.connector=quarkus-solace
-mp.messaging.outgoing.hello-out.producer.topic=
+mp.messaging.outgoing.hello-out.producer.topic=hello/foobar
+mp.messaging.outgoing.hello-out.merge=true
 
 mp.messaging.incoming.hello-in.connector=quarkus-solace
 mp.messaging.incoming.hello-in.consumer.queue.supports-nacks=true
-mp.messaging.incoming.hello-in.consumer.queue.name=
+mp.messaging.incoming.hello-in.consumer.queue.name=queue.foobar
+mp.messaging.incoming.hello-in.consumer.queue.missing-resource-creation-strategy=create-on-start
 mp.messaging.incoming.hello-in.consumer.queue.type=durable-exclusive
-#mp.messaging.incoming.hello-in.consumer.queue.failure-strategy=error_topic
-#mp.messaging.incoming.hello-in.consumer.queue.error.topic=
+mp.messaging.incoming.hello-in.consumer.queue.add-additional-subscriptions=true
+mp.messaging.incoming.hello-in.consumer.queue.subscriptions=hello/foobar
 
 mp.messaging.incoming.dynamic-destination-in.connector=quarkus-solace
 mp.messaging.incoming.dynamic-destination-in.consumer.queue.supports-nacks=true
-mp.messaging.incoming.dynamic-destination-in.consumer.queue.name=
+mp.messaging.incoming.dynamic-destination-in.consumer.queue.name=queue.dynamic.topic
+mp.messaging.incoming.dynamic-destination-in.consumer.queue.missing-resource-creation-strategy=create-on-start
 mp.messaging.incoming.dynamic-destination-in.consumer.queue.type=durable-exclusive
-mp.messaging.incoming.hello-in.consumer.queue.failure-strategy=ignore
-mp.messaging.incoming.dynamic-destination-in.consumer.queue.error.topic=solace/quarkus/error
+mp.messaging.incoming.dynamic-destination-in.consumer.queue.add-additional-subscriptions=true
+mp.messaging.incoming.dynamic-destination-in.consumer.queue.subscriptions=test/topic/>
 
 mp.messaging.outgoing.dynamic-destination-out.connector=quarkus-solace
-mp.messaging.outgoing.dynamic-destination-out.producer.topic=
+mp.messaging.outgoing.dynamic-destination-out.producer.topic=test/dynamic/topic