ARTEMIS-5801 Add a subscription name option to the consumer CLI tool

Adds option 'subscriptionName' to the consumer CLI options to allow
consuming from an existing durable subscription not created by the CLI

Author: tabish121

artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/Consumer.java

@@ -50,6 +50,9 @@ public class Consumer extends DestAbstract {
    @Option(names = "--data", description = "Serialize the messages to the specified file as they are consumed.")
    String file;
 
+   @Option(names = "--subscriptionName", description = "The subscription name to use for a durable consumer.")
+   String subscriptionName;
+
    @Override
    public Object execute(ActionContext context) throws Exception {
       super.execute(context);
@@ -78,6 +81,15 @@ public Object execute(ActionContext context) throws Exception {
          serializer.start();
       }
 
+      if (subscriptionName != null) {
+         if (threads != 1) {
+            context.err.println("Error: Cannot assign a subscription name when multiple threads are also configured.");
+            return null;
+         } else {
+            context.out.println("Consumer:: subscription name = " + subscriptionName);
+         }
+      }
+
       ConnectionFactory factory = createConnectionFactory();
 
       try (Connection connection = factory.createConnection()) {
@@ -92,7 +104,11 @@ public Object execute(ActionContext context) throws Exception {
             }
 
             Destination dest = getDestination(session);
-            threadsArray[i] = new ConsumerThread(session, dest, i, context);
+            if (subscriptionName == null) {
+               threadsArray[i] = new ConsumerThread(session, dest, i, context);
+            } else {
+               threadsArray[i] = new ConsumerThread(session, dest, subscriptionName, context);
+            }
 
             threadsArray[i]
                .setVerbose(verbose)
@@ -191,4 +207,13 @@ public Consumer setFile(String file) {
       this.file = file;
       return this;
    }
+
+   public String getSubscriptionName() {
+      return subscriptionName;
+   }
+
+   public Consumer setSubscriptionName(String subscriptionName) {
+      this.subscriptionName = subscriptionName;
+      return this;
+   }
 }

artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/ConsumerThread.java

@@ -57,7 +57,12 @@ public class ConsumerThread extends Thread {
    MessageListener listener;
 
    public ConsumerThread(Session session, Destination destination, int threadNr, ActionContext context) {
-      super("Consumer " + destination.toString() + ", thread=" + threadNr);
+      this(session, destination, "Consumer " + destination.toString() + ", thread=" + threadNr, context);
+   }
+
+   public ConsumerThread(Session session, Destination destination, String subscriptionName, ActionContext context) {
+      super(subscriptionName);
+
       this.destination = destination;
       this.session = session;
       this.context = context;

artemis-cli/src/test/java/org/apache/activemq/cli/test/CliConsumerTest.java

@@ -17,8 +17,9 @@
 package org.apache.activemq.cli.test;
 
 import javax.jms.Connection;
-
+import javax.jms.Session;
 import org.apache.activemq.artemis.api.core.Pair;
+import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.cli.commands.messages.Consumer;
 import org.apache.activemq.artemis.cli.commands.messages.Producer;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
@@ -29,6 +30,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class CliConsumerTest extends CliTestBase {
@@ -113,4 +115,50 @@ private void sendAndConsume(long messageCount, int timeout) throws Exception {
 
       Wait.assertEquals(0L, () -> server.locateQueue(address).getMessageCount(), 2000, 50);
    }
+
+   @Test
+   public void testConsumeFromExistingDurableSubscription() throws Exception {
+      final String address = "test-topic";
+      final String addressPrefix = "topic://";
+      final String clientID = "test-client";
+      final String subscriptionName = "test-sub";
+      final String credentials = "admin";
+      final TestActionContext context = new TestActionContext();
+
+      // Creates the durable subscription to consumer from using the CLI tool.
+      try (Connection connection = cf.createConnection(credentials, credentials)) {
+
+         connection.setClientID(clientID);
+         connection.start();
+
+         final Session session = createSession(connection);
+
+         session.createDurableConsumer(session.createTopic(address), subscriptionName);
+      }
+
+      produceMessages(addressPrefix + address, TEST_MESSAGE_COUNT);
+
+      server.addressQuery(SimpleString.of(address));
+
+      final String subscriptionQueueName = server.bindingQuery(SimpleString.of(address)).getQueueNames().get(0).toString();
+      assertNotNull(subscriptionQueueName);
+      final org.apache.activemq.artemis.core.server.Queue subscriptionQueue = server.locateQueue(subscriptionQueueName);
+      Wait.assertEquals((long) TEST_MESSAGE_COUNT, () -> subscriptionQueue.getMessageCount(), 2000, 50);
+
+      // Consume from the durable subscription with messages added.
+      new Consumer()
+         .setSubscriptionName(subscriptionName)
+         .setReceiveTimeout(100)
+         .setBreakOnNull(true)
+         .setDurable(true)
+         .setMessageCount(TEST_MESSAGE_COUNT)
+         .setDestination(addressPrefix + address)
+         .setClientID(clientID)
+         .setUser(credentials)
+         .setPassword(credentials)
+         .execute(context);
+
+      Wait.assertTrue(() -> context.getStdout().contains("subscription name"), 2000, 100);
+      Wait.assertEquals(0L, () -> subscriptionQueue.getMessageCount(), 2000, 50);
+   }
 }