You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/12/16 19:17:12 UTC
[activemq-artemis] branch master updated: ARTEMIS-3034 CLI Transfer
Option
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new 02bcb31 ARTEMIS-3034 CLI Transfer Option
new 95cb34c This closes #3383
02bcb31 is described below
commit 02bcb3195f2c7911750eedcff775837cb2af2977
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Mon Dec 14 17:33:05 2020 -0500
ARTEMIS-3034 CLI Transfer Option
---
.../org/apache/activemq/artemis/cli/Artemis.java | 3 +-
.../artemis/cli/commands/ActionAbstract.java | 43 +++
.../cli/commands/messages/ConnectionAbstract.java | 45 +--
.../artemis/cli/commands/messages/Transfer.java | 361 +++++++++++++++++++++
.../apache/activemq/artemis/util/ServerUtil.java | 24 +-
tests/smoke-tests/pom.xml | 32 ++
.../artemis/tests/smoke/transfer/TestTransfer.java | 332 +++++++++++++++++++
7 files changed, 791 insertions(+), 49 deletions(-)
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/Artemis.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/Artemis.java
index 0ceb29e..e5b908c 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/Artemis.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/Artemis.java
@@ -34,6 +34,7 @@ import org.apache.activemq.artemis.cli.commands.Mask;
import org.apache.activemq.artemis.cli.commands.check.HelpCheck;
import org.apache.activemq.artemis.cli.commands.check.NodeCheck;
import org.apache.activemq.artemis.cli.commands.check.QueueCheck;
+import org.apache.activemq.artemis.cli.commands.messages.Transfer;
import org.apache.activemq.artemis.cli.commands.queue.StatQueue;
import org.apache.activemq.artemis.cli.commands.Run;
import org.apache.activemq.artemis.cli.commands.Stop;
@@ -155,7 +156,7 @@ public class Artemis {
private static Cli.CliBuilder<Action> builder(File artemisInstance) {
String instance = artemisInstance != null ? artemisInstance.getAbsolutePath() : System.getProperty("artemis.instance");
- Cli.CliBuilder<Action> builder = Cli.<Action>builder("artemis").withDescription("ActiveMQ Artemis Command Line").withCommand(HelpAction.class).withCommand(Producer.class).withCommand(Consumer.class).withCommand(Browse.class).withCommand(Mask.class).withDefaultCommand(HelpAction.class);
+ Cli.CliBuilder<Action> builder = Cli.<Action>builder("artemis").withDescription("ActiveMQ Artemis Command Line").withCommand(HelpAction.class).withCommand(Producer.class).withCommand(Transfer.class).withCommand(Consumer.class).withCommand(Browse.class).withCommand(Mask.class).withDefaultCommand(HelpAction.class);
builder.withGroup("check").withDescription("Check tools group (node|queue) (example ./artemis check node)").
withDefaultCommand(HelpCheck.class).withCommands(NodeCheck.class, QueueCheck.class);
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/ActionAbstract.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/ActionAbstract.java
index 79563d3..914cbae 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/ActionAbstract.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/ActionAbstract.java
@@ -17,9 +17,17 @@
package org.apache.activemq.artemis.cli.commands;
import java.io.File;
+import java.net.InetAddress;
import java.net.URI;
+import java.util.Map;
import io.airlift.airline.Option;
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.core.config.FileDeploymentManager;
+import org.apache.activemq.artemis.core.config.impl.FileConfiguration;
+import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
+import org.apache.activemq.artemis.utils.ConfigurationHelper;
+import org.apache.activemq.artemis.utils.uri.SchemaConstants;
public abstract class ActionAbstract implements Action {
@@ -67,6 +75,41 @@ public abstract class ActionAbstract implements Action {
return brokerInstance;
}
+
+ protected String getBrokerURLInstance() {
+ if (getBrokerInstance() != null) {
+ try {
+ FileConfiguration fileConfiguration = new FileConfiguration();
+ String brokerConfiguration = new File(new File(getBrokerEtc()), "broker.xml").toURI().toASCIIString();
+ FileDeploymentManager fileDeploymentManager = new FileDeploymentManager(brokerConfiguration);
+ fileDeploymentManager.addDeployable(fileConfiguration);
+ fileDeploymentManager.readConfiguration();
+
+ for (TransportConfiguration acceptorConfiguration: fileConfiguration.getAcceptorConfigurations()) {
+ if (acceptorConfiguration.getName().equals("artemis")) {
+ Map<String, Object> acceptorParams = acceptorConfiguration.getParams();
+ String scheme = ConfigurationHelper.getStringProperty(TransportConstants.SCHEME_PROP_NAME, SchemaConstants.TCP, acceptorParams);
+ String host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, "localhost", acceptorParams);
+ int port = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, 61616, acceptorParams);
+
+ if (InetAddress.getByName(host).isAnyLocalAddress()) {
+ host = "localhost";
+ }
+
+ return new URI(scheme, null, host, port, null, null, null).toString();
+ }
+ }
+ } catch (Exception e) {
+ if (isVerbose()) {
+ System.out.print("Can not get the broker url instance: " + e.toString());
+ }
+ }
+ }
+
+ return null;
+ }
+
+
public String getBrokerEtc() {
if (brokerEtc == null) {
brokerEtc = System.getProperty("artemis.instance.etc");
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/ConnectionAbstract.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/ConnectionAbstract.java
index c11127d..74fb7ba 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/ConnectionAbstract.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/ConnectionAbstract.java
@@ -21,28 +21,18 @@ import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.JMSSecurityException;
-import java.io.File;
-import java.net.InetAddress;
-import java.net.URI;
-import java.util.Map;
import io.airlift.airline.Option;
-import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.cli.commands.ActionContext;
import org.apache.activemq.artemis.cli.commands.InputAbstract;
-import org.apache.activemq.artemis.core.config.FileDeploymentManager;
-import org.apache.activemq.artemis.core.config.impl.FileConfiguration;
-import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
-import org.apache.activemq.artemis.utils.ConfigurationHelper;
-import org.apache.activemq.artemis.utils.uri.SchemaConstants;
import org.apache.qpid.jms.JmsConnectionFactory;
public class ConnectionAbstract extends InputAbstract {
private static final String DEFAULT_BROKER_URL = "tcp://localhost:61616";
- @Option(name = "--url", description = "URL towards the broker. (default: tcp://localhost:61616)")
+ @Option(name = "--source-url", description = "URL towards the broker. (default: Read from current broker.xml or tcp://localhost:61616 if the default cannot be parsed)")
protected String brokerURL = DEFAULT_BROKER_URL;
@Option(name = "--user", description = "User used to connect")
@@ -117,39 +107,6 @@ public class ConnectionAbstract extends InputAbstract {
return null;
}
- private String getBrokerURLInstance() {
- if (getBrokerInstance() != null) {
- try {
- FileConfiguration fileConfiguration = new FileConfiguration();
- String brokerConfiguration = new File(new File(getBrokerEtc()), "broker.xml").toURI().toASCIIString();
- FileDeploymentManager fileDeploymentManager = new FileDeploymentManager(brokerConfiguration);
- fileDeploymentManager.addDeployable(fileConfiguration);
- fileDeploymentManager.readConfiguration();
-
- for (TransportConfiguration acceptorConfiguration: fileConfiguration.getAcceptorConfigurations()) {
- if (acceptorConfiguration.getName().equals("artemis")) {
- Map<String, Object> acceptorParams = acceptorConfiguration.getParams();
- String scheme = ConfigurationHelper.getStringProperty(TransportConstants.SCHEME_PROP_NAME, SchemaConstants.TCP, acceptorParams);
- String host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, "localhost", acceptorParams);
- int port = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, 61616, acceptorParams);
-
- if (InetAddress.getByName(host).isAnyLocalAddress()) {
- host = "localhost";
- }
-
- return new URI(scheme, null, host, port, null, null, null).toString();
- }
- }
- } catch (Exception e) {
- if (isVerbose()) {
- System.out.print("Can not get the broker url instance: " + e.toString());
- }
- }
- }
-
- return null;
- }
-
protected ConnectionFactory createConnectionFactory() throws Exception {
if (protocol.equals("core")) {
return createCoreConnectionFactory();
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/Transfer.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/Transfer.java
new file mode 100644
index 0000000..7f7b7d0
--- /dev/null
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/messages/Transfer.java
@@ -0,0 +1,361 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.cli.commands.messages;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.JMSSecurityException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import io.airlift.airline.Command;
+import io.airlift.airline.Option;
+import org.apache.activemq.artemis.api.core.Pair;
+import org.apache.activemq.artemis.cli.commands.ActionContext;
+import org.apache.activemq.artemis.cli.commands.InputAbstract;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.qpid.jms.JmsConnectionFactory;
+
+@Command(name = "transfer", description = "Moves Messages from one destination towards another destination")
+public class Transfer extends InputAbstract {
+
+ private static final String DEFAULT_BROKER_URL = "tcp://localhost:61616";
+
+ @Option(name = "--source-url", description = "URL towards the broker. (default: Read from current broker.xml or tcp://localhost:61616 if the default cannot be parsed)")
+ protected String sourceURL = DEFAULT_BROKER_URL;
+
+ @Option(name = "--source-user", description = "User used to connect")
+ protected String sourceUser;
+
+ @Option(name = "--source-password", description = "Password used to connect")
+ protected String sourcePassword;
+
+ @Option(name = "--target-url", description = "URL towards the broker. (default: Read from current broker.xml or tcp://localhost:61616 if the default cannot be parsed)")
+ protected String targetURL = DEFAULT_BROKER_URL;
+
+ @Option(name = "--target-user", description = "User used to connect")
+ protected String targetUser;
+
+ @Option(name = "--target-password", description = "Password used to connect")
+ protected String targetPassword;
+
+ @Option(name = "--receive-timeout", description = "Amount of time (in milliseconds) to wait before giving up the loop. 0 means receiveNoWait, -1 means consumer.receive() waiting forever. (default=5000)")
+ int receiveTimeout = 5000;
+
+ @Option(name = "--source-client-id", description = "ClientID to be associated with connection")
+ String sourceClientID;
+
+ @Option(name = "--source-protocol", description = "Protocol used. Valid values are amqp or core. Default=core.")
+ String sourceProtocol = "core";
+
+ @Option(name = "--source-queue", description = "JMS Queue to be used.")
+ String sourceQueue;
+
+ @Option(name = "--shared-durable-subscription", description = "Name of a shared subscription name to be used on the input topic")
+ String sharedDurableSubscription;
+
+ @Option(name = "--shared-subscription", description = "Name of a shared subscription name to be used on the input topic")
+ String sharedSubscription;
+
+ @Option(name = "--durable-consumer", description = "Name of a durable consumer to be used on the input topic")
+ String durableConsumer;
+
+ @Option(name = "--no-Local", description = "Use noLocal when applicable on topic operation")
+ boolean noLocal;
+
+ @Option(name = "--source-topic", description = "Destination to be used. It can be prefixed with queue:// or topic:// and can be an FQQN in the form of <address>::<queue>. (Default: queue://TEST)")
+ String sourceTopic;
+
+ @Option(name = "--source-filter", description = "filter to be used with the consumer")
+ String filter;
+
+ @Option(name = "--target-protocol", description = "Protocol used. Valid values are amqp or core. Default=core.")
+ String targetProtocol = "core";
+
+ @Option(name = "--commit-interval", description = "Destination to be used. It can be prefixed with queue:// or topic:// and can be an FQQN in the form of <address>::<queue>. (Default: queue://TEST)")
+ int commitInterval = 1000;
+
+ @Option(name = "--copy", description = "If this option is chosen we will perform a copy of the queue by rolling back the original TX on the source.")
+ boolean copy;
+
+ @Option(name = "--target-queue", description = "JMS Queue to be used.")
+ String targetQueue;
+
+ @Option(name = "--target-topic", description = "Destination to be used. It can be prefixed with queue:// or topic:// and can be an FQQN in the form of <address>::<queue>. (Default: queue://TEST)")
+ String targetTopic;
+
+ boolean isCopy() {
+ return copy;
+ }
+
+ @SuppressWarnings("StringEquality")
+ @Override
+ public Object execute(ActionContext context) throws Exception {
+ super.execute(context);
+
+ // it is intentional to make a comparison on the String object here
+ // this is to test if the original option was switched or not.
+ // we don't care about being .equals at all.
+ // as a matter of fact if you pass brokerURL in a way it's equals to DEFAULT_BROKER_URL,
+ // we should not the broker URL Instance
+ // and still honor the one passed by parameter.
+ // SupressWarnings was added to this method to supress the false positive here from error-prone.
+ if (sourceURL == DEFAULT_BROKER_URL) {
+ String brokerURLInstance = getBrokerURLInstance();
+
+ if (brokerURLInstance != null) {
+ sourceURL = brokerURLInstance;
+ }
+ }
+
+ System.out.println("Connection brokerURL = " + sourceURL);
+
+ ConnectionFactory sourceConnectionFactory = createConnectionFactory("source", sourceProtocol, sourceURL, sourceUser, sourcePassword, sourceClientID);
+ Connection sourceConnection = sourceConnectionFactory.createConnection();
+
+ Session sourceSession = sourceConnection.createSession(Session.SESSION_TRANSACTED);
+ Destination sourceDestination = createDestination("source", sourceSession, sourceQueue, sourceTopic);
+ MessageConsumer consumer = null;
+ if (sourceDestination instanceof Queue) {
+ if (filter != null) {
+ consumer = sourceSession.createConsumer(sourceDestination, filter);
+ } else {
+ consumer = sourceSession.createConsumer(sourceDestination);
+ }
+ } else if (sourceDestination instanceof Topic) {
+
+ Topic topic = (Topic) sourceDestination;
+
+ if (durableConsumer != null) {
+ if (filter != null) {
+ consumer = sourceSession.createDurableConsumer(topic, durableConsumer);
+ } else {
+ consumer = sourceSession.createDurableConsumer(topic, durableConsumer, filter, noLocal);
+ }
+ } else if (sharedDurableSubscription != null) {
+ if (filter != null) {
+ consumer = sourceSession.createSharedDurableConsumer(topic, sharedDurableSubscription, filter);
+ } else {
+ consumer = sourceSession.createSharedDurableConsumer(topic, sharedDurableSubscription);
+ }
+ } else if (sharedSubscription != null) {
+ if (filter != null) {
+ consumer = sourceSession.createSharedConsumer(topic, sharedSubscription, filter);
+ } else {
+ consumer = sourceSession.createSharedConsumer(topic, sharedSubscription);
+ }
+ } else {
+ throw new IllegalArgumentException("you have to specify --durable-consumer, --shared-durable-subscription or --shared-subscription with a topic");
+ }
+ }
+
+ ConnectionFactory targetConnectionFactory = createConnectionFactory("target", targetProtocol, targetURL, targetUser, targetPassword, null);
+ Connection targetConnection = targetConnectionFactory.createConnection();
+ Session targetSession = targetConnection.createSession(Session.SESSION_TRANSACTED);
+ Destination targetDestination = createDestination("target", targetSession, targetQueue, targetTopic);
+ MessageProducer producer = targetSession.createProducer(targetDestination);
+
+ if (sourceURL.equals(targetURL) && sourceDestination.equals(targetDestination)) {
+ System.out.println("You cannot transfer between " + sourceURL + "/" + sourceDestination + " and " + targetURL + "/" + targetDestination + ".\n" + "That would create an infinite recursion.");
+ throw new IllegalArgumentException("cannot use " + sourceDestination + " == " + targetDestination);
+ }
+
+ sourceConnection.start();
+ int pending = 0, total = 0;
+ while (true) {
+
+ Message receivedMessage;
+ if (receiveTimeout < 0) {
+ receivedMessage = consumer.receive();
+ } else if (receiveTimeout == 0) {
+ receivedMessage = consumer.receiveNoWait();
+ } else {
+ receivedMessage = consumer.receive(receiveTimeout);
+ }
+
+ if (receivedMessage == null) {
+ if (isVerbose()) {
+ System.out.println("could not receive any more messages");
+ }
+ break;
+ }
+ producer.send(receivedMessage);
+ pending++;
+ total++;
+
+ if (isVerbose()) {
+ System.out.println("Received message " + total + " with " + pending + " messages pending to be commited");
+ }
+ if (pending > commitInterval) {
+ System.out.println("Transferred " + pending + " messages of " + total);
+ pending = 0;
+ targetSession.commit();
+ if (!isCopy()) {
+ sourceSession.commit();
+ }
+ }
+ }
+
+ System.out.println("Transferred a total of " + total + " messages");
+
+ if (pending != 0) {
+ targetSession.commit();
+ if (isCopy()) {
+ sourceSession.rollback();
+ } else {
+ sourceSession.commit();
+ }
+ }
+
+ sourceConnection.close();
+ targetConnection.close();
+
+ return null;
+ }
+
+ Destination createDestination(String role, Session session, String queue, String topic) throws Exception {
+ if (queue != null && topic != null) {
+ throw new IllegalArgumentException("Cannot have topic and queue passed as " + role);
+ }
+
+ if (queue != null) {
+ return session.createQueue(queue);
+ }
+
+ if (topic != null) {
+ return session.createTopic(topic);
+ }
+
+ throw new IllegalArgumentException("You need to pass either a topic or a queue as " + role);
+ }
+
+ protected ConnectionFactory createConnectionFactory(String role,
+ String protocol,
+ String brokerURL,
+ String user,
+ String password,
+ String clientID) throws Exception {
+ if (protocol.equals("core")) {
+ if (isVerbose()) {
+ System.out.println("Creating " + role + " CORE Connection towards " + brokerURL);
+ }
+ return createCoreConnectionFactory(brokerURL, user, password, clientID);
+ } else if (protocol.equals("amqp")) {
+ if (isVerbose()) {
+ System.out.println("Creating " + role + " AMQP Connection towards " + brokerURL);
+ }
+ return createAMQPConnectionFactory(brokerURL, user, password, clientID);
+ } else {
+ throw new IllegalStateException("protocol " + protocol + " not supported");
+ }
+ }
+
+ private ConnectionFactory createAMQPConnectionFactory(String brokerURL,
+ String user,
+ String password,
+ String clientID) {
+ if (brokerURL.startsWith("tcp://")) {
+ // replacing tcp:// by amqp://
+ brokerURL = "amqp" + brokerURL.substring(3);
+ }
+ JmsConnectionFactory cf = new JmsConnectionFactory(user, password, brokerURL);
+ if (clientID != null) {
+ cf.setClientID(clientID);
+ }
+
+ try {
+ Connection connection = cf.createConnection();
+ connection.close();
+ return cf;
+ } catch (JMSSecurityException e) {
+ // if a security exception will get the user and password through an input
+ context.err.println("Connection failed::" + e.getMessage());
+ userPassword(brokerURL);
+ cf = new JmsConnectionFactory(user, password, brokerURL);
+ if (clientID != null) {
+ cf.setClientID(clientID);
+ }
+ return cf;
+ } catch (JMSException e) {
+ // if a connection exception will ask for the URL, user and password
+ context.err.println("Connection failed::" + e.getMessage());
+ brokerURL = input("--url", "Type in the broker URL for a retry (e.g. tcp://localhost:61616)", brokerURL);
+ userPassword(brokerURL);
+ cf = new JmsConnectionFactory(user, password, brokerURL);
+ if (clientID != null) {
+ cf.setClientID(clientID);
+ }
+ return cf;
+ }
+ }
+
+ protected ActiveMQConnectionFactory createCoreConnectionFactory(String brokerURL,
+ String user,
+ String password,
+ String clientID) {
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(brokerURL, user, password);
+
+ if (clientID != null) {
+ System.out.println("Consumer:: clientID = " + clientID);
+ cf.setClientID(clientID);
+ }
+ try {
+ Connection connection = cf.createConnection();
+ connection.close();
+ return cf;
+ } catch (JMSSecurityException e) {
+ // if a security exception will get the user and password through an input
+ if (context != null) {
+ context.err.println("Connection failed::" + e.getMessage());
+ }
+ Pair<String, String> userPair = userPassword(brokerURL);
+ cf = new ActiveMQConnectionFactory(brokerURL, userPair.getA(), userPair.getB());
+ if (clientID != null) {
+ cf.setClientID(clientID);
+ }
+ return cf;
+ } catch (JMSException e) {
+ // if a connection exception will ask for the URL, user and password
+ if (context != null) {
+ context.err.println("Connection failed::" + e.getMessage());
+ }
+ brokerURL = input("--url", "Type in the broker URL for a retry (e.g. tcp://localhost:61616)", brokerURL);
+ Pair<String, String> userPair = userPassword(brokerURL);
+ cf = new ActiveMQConnectionFactory(brokerURL, userPair.getA(), userPair.getB());
+ if (clientID != null) {
+ cf.setClientID(clientID);
+ }
+ return cf;
+ }
+ }
+
+ Pair<String, String> userPassword(String uri) {
+ System.out.println("Type in user/password towards " + uri);
+ String user, password;
+ user = input("--user", "Type the username for a retry", null);
+ password = inputPassword("--password", "Type the password for a retry", null);
+ return new Pair<>(user, password);
+ }
+
+}
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java
index 6d7cf8f..a953161 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java
@@ -23,6 +23,7 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
+import java.util.ArrayList;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.client.ClientSession;
@@ -71,16 +72,31 @@ public class ServerUtil {
private static Process internalStartServer(String artemisInstance,
String serverName) throws IOException, ClassNotFoundException {
+
+ return execute(artemisInstance, serverName, "run");
+ }
+
+ public static Process execute(String artemisInstance, String jobName, String...args) throws IOException, ClassNotFoundException {
try {
boolean IS_WINDOWS = System.getProperty("os.name").toLowerCase().trim().startsWith("win");
+ ArrayList<String> command = new ArrayList<>();
+
ProcessBuilder builder = null;
if (IS_WINDOWS) {
- builder = new ProcessBuilder("cmd", "/c", "artemis.cmd", "run");
+ command.add("cmd");
+ command.add("/c");
+ command.add("artemis.cmd");
} else {
- builder = new ProcessBuilder("./artemis", "run");
+ command.add("./artemis");
}
+ for (String arg: args) {
+ command.add(arg);
+ }
+
+ builder = new ProcessBuilder(command);
+
builder.directory(new File(artemisInstance + "/bin"));
final Process process = builder.start();
@@ -91,12 +107,12 @@ public class ServerUtil {
}
});
- ProcessLogger outputLogger = new ProcessLogger(true, process.getInputStream(), serverName, false);
+ ProcessLogger outputLogger = new ProcessLogger(true, process.getInputStream(), jobName, false);
outputLogger.start();
// Adding a reader to System.err, so the VM won't hang on a System.err.println as identified on this forum thread:
// http://www.jboss.org/index.html?module=bb&op=viewtopic&t=151815
- ProcessLogger errorLogger = new ProcessLogger(true, process.getErrorStream(), serverName, true);
+ ProcessLogger errorLogger = new ProcessLogger(true, process.getErrorStream(), jobName, true);
errorLogger.start();
return process;
} catch (IOException e) {
diff --git a/tests/smoke-tests/pom.xml b/tests/smoke-tests/pom.xml
index 3c5bd43..48f7f81 100644
--- a/tests/smoke-tests/pom.xml
+++ b/tests/smoke-tests/pom.xml
@@ -608,6 +608,38 @@
</args>
</configuration>
</execution>
+ <!-- used on TransferTest -->
+ <execution>
+ <phase>test-compile</phase>
+ <id>create-transfer-1</id>
+ <goals>
+ <goal>create</goal>
+ </goals>
+ <configuration>
+ <!-- this makes it easier in certain envs -->
+ <allowAnonymous>true</allowAnonymous>
+ <user>admin</user>
+ <password>admin</password>
+ <noWeb>true</noWeb>
+ <instance>${basedir}/target/transfer1</instance>
+ </configuration>
+ </execution>
+ <execution>
+ <phase>test-compile</phase>
+ <id>create-transfer-2</id>
+ <goals>
+ <goal>create</goal>
+ </goals>
+ <configuration>
+ <!-- this makes it easier in certain envs -->
+ <allowAnonymous>true</allowAnonymous>
+ <user>admin</user>
+ <password>admin</password>
+ <noWeb>true</noWeb>
+ <portOffset>100</portOffset>
+ <instance>${basedir}/target/transfer2</instance>
+ </configuration>
+ </execution>
</executions>
<dependencies>
<dependency>
diff --git a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/transfer/TestTransfer.java b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/transfer/TestTransfer.java
new file mode 100644
index 0000000..1c4b235
--- /dev/null
+++ b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/transfer/TestTransfer.java
@@ -0,0 +1,332 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.tests.smoke.transfer;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.qpid.jms.JmsConnectionFactory;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestTransfer extends SmokeTestBase {
+
+ public static final String SERVER_NAME_0 = "transfer1";
+ public static final String SERVER_NAME_1 = "transfer2";
+ private static final int NUMBER_OF_MESSAGES = 200;
+ private static final int PARTIAL_MESSAGES = 10;
+ String sourceTransferProtocol = "amqp";
+ String targetTransferProtocol = "amqp";
+ String senderProtocol = "amqp";
+ String consumerProtocol = "amqp";
+
+ public TestTransfer(String sender, String consumer, String source, String target) {
+ this.senderProtocol = sender;
+ this.consumerProtocol = consumer;
+ this.sourceTransferProtocol = source;
+ this.targetTransferProtocol = target;
+ }
+
+ @Parameterized.Parameters(name = "sender={0}, consumer={1}, sourceOnTransfer={2}, targetOnTransfer={3}")
+ public static Collection<Object[]> getParams() {
+
+ String[] protocols = new String[]{"core", "amqp"};
+
+ ArrayList<Object[]> parameters = new ArrayList<>();
+
+ for (int i = 0; i < protocols.length; i++) {
+ for (int j = 0; j < protocols.length; j++) {
+ // sender and sourceOnTransfer have to be the same
+ // consumer and targetOnTransfer have to be the same
+ // this is because AMQP Shared Subscription will create a different queue than core
+ String[] parameter = new String[]{protocols[i], protocols[j], protocols[i], protocols[j]};
+ parameters.add(parameter);
+ }
+ }
+
+ return parameters;
+ }
+
+ private static ConnectionFactory createConnectionFactory(String protocol, String uri) {
+ if (protocol.toUpperCase().equals("AMQP")) {
+
+ if (uri.startsWith("tcp://")) {
+ // replacing tcp:// by amqp://
+ uri = "amqp" + uri.substring(3);
+
+ }
+ return new JmsConnectionFactory(uri);
+ } else if (protocol.toUpperCase().equals("CORE") || protocol.toUpperCase().equals("ARTEMIS")) {
+ return new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(uri);
+ } else {
+ throw new IllegalStateException("Unkown:" + protocol);
+ }
+ }
+
+ private ConnectionFactory createConsumerCF() {
+ return createConnectionFactory(consumerProtocol, "tcp://localhost:61716");
+ }
+
+ private ConnectionFactory createSenderCF() {
+ return createConnectionFactory(senderProtocol, "tcp://localhost:61616");
+ }
+
+ @Before
+ public void before() throws Exception {
+ cleanupData(SERVER_NAME_0);
+ cleanupData(SERVER_NAME_1);
+ disableCheckThread();
+ startServer(SERVER_NAME_0, 0, 30000);
+ startServer(SERVER_NAME_1, 100, 30000);
+ }
+
+ @Test
+ public void testTransferSimpleQueueCopy() throws Exception {
+ internalTransferSimpleQueue(false);
+ }
+
+ @Test
+ public void testTransferSimpleQueue() throws Exception {
+ internalTransferSimpleQueue(true);
+ }
+
+ public String getQueueName() {
+ return getName();
+ }
+
+ public String getTopicName() {
+ return "Topic" + getName();
+ }
+
+ private void internalTransferSimpleQueue(boolean copy) throws Exception {
+ ConnectionFactory factory = createSenderCF();
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+
+ Queue queue = session.createQueue(getQueueName());
+ MessageProducer producer = session.createProducer(queue);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ producer.send(session.createTextMessage("hello " + i));
+ }
+
+ session.commit();
+
+ String[] argsArray = new String[]{"transfer", "--target-url", "tcp://localhost:61716", "--source-queue", getQueueName(), "--target-queue", getQueueName(), "--source-protocol", sourceTransferProtocol, "--target-protocol", targetTransferProtocol, "--receive-timeout", "0"};
+
+ if (copy) {
+ ArrayList<String> copyArgs = new ArrayList<>();
+ for (String a : argsArray) {
+ copyArgs.add(a);
+ }
+ if (copy) {
+ copyArgs.add("--copy");
+ }
+ argsArray = copyArgs.toArray(new String[copyArgs.size()]);
+ }
+
+ Process transferProcess = ServerUtil.execute(getServerLocation(SERVER_NAME_0), "transfer", argsArray);
+ transferProcess.waitFor();
+
+ ConnectionFactory factoryTarget = createConsumerCF();
+ Connection connectionTarget = factoryTarget.createConnection();
+ connectionTarget.start();
+ Session sessionTarget = connectionTarget.createSession(true, Session.SESSION_TRANSACTED);
+
+ Queue queueTarget = sessionTarget.createQueue(getQueueName());
+ MessageConsumer consumer = sessionTarget.createConsumer(queueTarget);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ TextMessage received = (TextMessage) consumer.receive(1000);
+ Assert.assertNotNull(received);
+ Assert.assertEquals("hello " + i, received.getText());
+ }
+
+ sessionTarget.commit();
+
+ Assert.assertNull(consumer.receiveNoWait());
+
+ MessageConsumer consumerSource = session.createConsumer(queue);
+ connection.start();
+
+ if (copy) {
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ TextMessage received = (TextMessage) consumerSource.receive(1000);
+ Assert.assertNotNull(received);
+ Assert.assertEquals("hello " + i, received.getText());
+ }
+ }
+
+ Assert.assertNull(consumerSource.receiveNoWait());
+
+ connection.close();
+ connectionTarget.close();
+ }
+
+ @Test
+ public void testDurableSharedSubscrition() throws Exception {
+ ConnectionFactory factory = createSenderCF();
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+
+ Topic topic = session.createTopic(getTopicName());
+ MessageConsumer subscription = session.createSharedDurableConsumer(topic, "testSubs");
+ MessageProducer producer = session.createProducer(topic);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ producer.send(session.createTextMessage("hello " + i));
+ }
+
+ session.commit();
+ subscription.close();
+
+ Process transferProcess = ServerUtil.execute(getServerLocation(SERVER_NAME_0), "transfer", "transfer", "--target-url", "tcp://localhost:61716", "--source-topic", getTopicName(), "--shared-durable-subscription", "testSubs", "--target-queue", getQueueName(), "--source-protocol", sourceTransferProtocol, "--target-protocol", targetTransferProtocol, "--receive-timeout", "1000", "--verbose");
+ transferProcess.waitFor();
+
+ ConnectionFactory factoryTarget = createConsumerCF();
+ Connection connectionTarget = factoryTarget.createConnection();
+ connectionTarget.start();
+ Session sessionTarget = connectionTarget.createSession(true, Session.SESSION_TRANSACTED);
+
+ Queue queueTarget = sessionTarget.createQueue(getQueueName());
+ MessageConsumer consumer = sessionTarget.createConsumer(queueTarget);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ TextMessage received = (TextMessage) consumer.receive(1000);
+ Assert.assertNotNull(received);
+ }
+
+ sessionTarget.commit();
+
+ Assert.assertNull(consumer.receiveNoWait());
+
+ connection.close();
+ connectionTarget.close();
+ }
+
+ @Test
+ public void testSharedSubscrition() throws Exception {
+ ConnectionFactory factory = createSenderCF();
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+
+ Topic topic = session.createTopic(getTopicName());
+ MessageConsumer subscription = session.createSharedConsumer(topic, "testSubs");
+ MessageProducer producer = session.createProducer(topic);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ // I can't just send a few messages.. I have to send more messages that would be stuck on delivering mode
+ // since the transfer will share the consumer with the shared consumer
+ // and the temporary queue would be removed if i closed it earlier.
+ for (int i = 0; i < 2000; i++) {
+ producer.send(session.createTextMessage("hello " + i));
+ }
+
+ session.commit();
+
+ Process transferProcess = ServerUtil.execute(getServerLocation(SERVER_NAME_0), "transfer", "transfer", "--target-url", "tcp://localhost:61716", "--source-topic", getTopicName(), "--shared-subscription", "testSubs", "--target-queue", getQueueName(), "--source-protocol", sourceTransferProtocol, "--target-protocol", targetTransferProtocol, "--receive-timeout", "0", "--verbose");
+ transferProcess.waitFor();
+
+ // this test is a bit tricky as the subscription would be removed when the consumer is gone...
+ // I'm adding a test for completion only
+ // and the subscription has to be closed only after the transfer,
+ // which will not receive all the messages as some messages will be in delivering mode
+ subscription.close();
+
+ ConnectionFactory factoryTarget = createConsumerCF();
+ Connection connectionTarget = factoryTarget.createConnection();
+ connectionTarget.start();
+ Session sessionTarget = connectionTarget.createSession(true, Session.SESSION_TRANSACTED);
+
+ Queue queueTarget = sessionTarget.createQueue(getQueueName());
+ MessageConsumer consumer = sessionTarget.createConsumer(queueTarget);
+
+ // we are keeping a non durable subscription so the temporary queue still up
+ // I'm not going to bother about being too strict about the content, just that some messages arrived
+ for (int i = 0; i < PARTIAL_MESSAGES; i++) {
+ TextMessage received = (TextMessage) consumer.receive(1000);
+ Assert.assertNotNull(received);
+ }
+
+ sessionTarget.commit();
+
+ connection.close();
+ connectionTarget.close();
+ }
+
+ @Test
+ public void testDurableConsumer() throws Exception {
+ ConnectionFactory factory = createSenderCF();
+ Connection connection = factory.createConnection();
+ connection.setClientID("test");
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+
+ Topic topic = session.createTopic(getTopicName());
+ MessageConsumer subscription = session.createDurableConsumer(topic, "testSubs");
+ MessageProducer producer = session.createProducer(topic);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ producer.send(session.createTextMessage("hello " + i));
+ }
+
+ session.commit();
+ connection.close();
+ subscription.close();
+
+ Process transferProcess = ServerUtil.execute(getServerLocation(SERVER_NAME_0), "transfer", "transfer", "--target-url", "tcp://localhost:61716", "--source-topic", getTopicName(), "--source-client-id", "test", "--durable-consumer", "testSubs", "--target-queue", getQueueName(), "--source-protocol", sourceTransferProtocol, "--target-protocol", targetTransferProtocol, "--receive-timeout", "1000", "--verbose", "--silent");
+ transferProcess.waitFor();
+
+ ConnectionFactory factoryTarget = createConsumerCF();
+ Connection connectionTarget = factoryTarget.createConnection();
+ connectionTarget.start();
+ Session sessionTarget = connectionTarget.createSession(true, Session.SESSION_TRANSACTED);
+
+ Queue queueTarget = sessionTarget.createQueue(getQueueName());
+ MessageConsumer consumer = sessionTarget.createConsumer(queueTarget);
+
+ // we are keeping a non durable subscription so the temporary queue still up
+ // I'm not going to bother about being too strict about the content, just that some messages arrived
+ for (int i = 0; i < PARTIAL_MESSAGES; i++) {
+ TextMessage received = (TextMessage) consumer.receive(1000);
+ Assert.assertNotNull(received);
+ }
+
+ sessionTarget.commit();
+
+ connectionTarget.close();
+ }
+
+}