com.fasterxml.jackson.core
jackson-databind
diff --git a/jmslib/src/main/java/com/redhat/mqe/lib/Content.java b/lib/src/main/java/com/redhat/mqe/lib/Content.java
similarity index 88%
rename from jmslib/src/main/java/com/redhat/mqe/lib/Content.java
rename to lib/src/main/java/com/redhat/mqe/lib/Content.java
index cb1875c4..0b906492 100644
--- a/jmslib/src/main/java/com/redhat/mqe/lib/Content.java
+++ b/lib/src/main/java/com/redhat/mqe/lib/Content.java
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017 Red Hat, Inc.
+ * Copyright (c) 2021 Red Hat, Inc.
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -36,7 +36,7 @@ public class Content {
private String key;
private Object value;
private Class type;
- private boolean isMap;
+ private final boolean isMap;
/**
* Create content from defined rules.
@@ -65,7 +65,7 @@ public Content(String contentType, String parsedValue, boolean isMap) {
splitValue = parsedValue.substring(parsedValue.indexOf(splitter) + 1);
}
} else if (parsedValue.contains("=")) {
- // last argument 'allowExplicityRetype' can be omitted as parsedValue will not by autotypecasted - no '~'
+ // last argument 'allowExplicitlyRetype' can be omitted as parsedValue will not by autotypecasted - no '~'
splitValue = parsedValue.substring(parsedValue.indexOf(splitter) + 1);
} else {
splitter = "~";
@@ -76,7 +76,11 @@ public Content(String contentType, String parsedValue, boolean isMap) {
this.key = parsedValue.substring(0, parsedValue.indexOf(splitter));
val = parsedValue.substring(parsedValue.indexOf(splitter) + 1);
} else {
- if (parsedValue.startsWith("~~")) {
+ if (parsedValue == null) {
+ this.type = void.class;
+ this.value = null;
+ return;
+ } else if (parsedValue.startsWith("~~")) {
contentType = "String";
}
this.type = Utils.getClassType(contentType, parsedValue, true);
@@ -87,12 +91,15 @@ public Content(String contentType, String parsedValue, boolean isMap) {
} catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
e.printStackTrace();
}
- } catch (JmsMessagingException e) {
+ } catch (MessagingException e) {
e.printStackTrace();
}
}
public String getKey() {
+ if (!isMap) {
+ throw new IllegalStateException("Only maps have keys");
+ }
return key;
}
diff --git a/lib/src/main/java/com/redhat/mqe/lib/LogConfigurator.java b/lib/src/main/java/com/redhat/mqe/lib/LogConfigurator.java
new file mode 100644
index 00000000..06a1fa55
--- /dev/null
+++ b/lib/src/main/java/com/redhat/mqe/lib/LogConfigurator.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2022 Red Hat, Inc.
+ *
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 com.redhat.mqe.lib;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+
+import java.util.function.Consumer;
+
+/**
+ * Utility class to configure log messages
+ *
+ * ...
+ * ...
+ */
+public class LogConfigurator {
+ /**
+ * Restricted constructor
+ */
+ private LogConfigurator() {
+ }
+
+
+ private static void configureCommon(Consumer customizeConfig) {
+ final LoggerContext context = LoggerContext.getContext(false);
+ final Configuration config = context.getConfiguration();
+ LoggerConfig rootLogger = config.getRootLogger();
+
+ customizeConfig.accept(rootLogger);
+
+ context.updateLoggers();
+ }
+
+ public static void trace() {
+ configureCommon((LoggerConfig config) -> config.setLevel(Level.TRACE));
+ }
+
+ /**
+ * Configure the output to be at debug level
+ */
+ public static void debug() {
+ configureCommon((LoggerConfig config) -> config.setLevel(Level.DEBUG));
+ }
+
+ /**
+ * Configure the output to be at info (info) level
+ */
+ public static void info() {
+ configureCommon((LoggerConfig config) -> config.setLevel(Level.INFO));
+ }
+
+ /**
+ * Configure the output to be as error as possible
+ */
+ public static void error() {
+ configureCommon((LoggerConfig config) -> config.setLevel(Level.ERROR));
+ }
+}
diff --git a/lib/src/main/java/com/redhat/mqe/lib/MessageFormatter.java b/lib/src/main/java/com/redhat/mqe/lib/MessageFormatter.java
index e785db3f..1154c175 100644
--- a/lib/src/main/java/com/redhat/mqe/lib/MessageFormatter.java
+++ b/lib/src/main/java/com/redhat/mqe/lib/MessageFormatter.java
@@ -24,7 +24,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.UnsupportedEncodingException;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
@@ -37,7 +40,7 @@
* JSON, and so on.
*/
public abstract class MessageFormatter {
- static Logger LOG = LoggerFactory.getLogger(MessageFormatter.class);
+ static final Logger LOG = LoggerFactory.getLogger(MessageFormatter.class);
private final ObjectMapper json = new ObjectMapper();
protected StringBuilder formatBool(Boolean in_data) {
@@ -113,6 +116,9 @@ protected StringBuilder formatDouble(double in_data) {
return int_res;
}
+ /**
+ Formats object as Python
+ */
@SuppressWarnings("unchecked")
protected StringBuilder formatObject(Object in_data) {
StringBuilder int_res = new StringBuilder();
@@ -135,14 +141,8 @@ protected StringBuilder formatObject(Object in_data) {
} else if (in_data instanceof UUID) {
int_res.append(formatString(in_data.toString()));
} else if (in_data instanceof byte[]) {
- try {
- String value = new String((byte[]) in_data, "UTF-8");
- int_res.append(formatString(value));
- } catch (UnsupportedEncodingException uee) {
- LOG.error("Error while getting message properties!", uee.getMessage());
- uee.printStackTrace();
- System.exit(1);
- }
+ String value = new String((byte[]) in_data, StandardCharsets.UTF_8);
+ int_res.append(formatString(value));
} else {
handleUnsupportedObjectMessagePayloadType(int_res, in_data);
}
@@ -238,4 +238,18 @@ public void printStatistics(Hashtable msg) {
public void printConnectorStatistics(int connectionsOpened, int connectionsFailed, int connectionsTotal) {
System.out.println(connectionsOpened + " " + connectionsFailed + " " + connectionsTotal);
}
+
+ public static String hash(Object o) {
+ if (o == null) {
+ return null; // no point in hashing this value
+ }
+ MessageDigest md;
+ try {
+ md = MessageDigest.getInstance("SHA-1");
+ } catch (NoSuchAlgorithmException e) {
+ throw new MessagingException("Unable to hash message", e);
+ }
+ String content = o.toString();
+ return new BigInteger(1, md.digest(content.getBytes())).toString(16);
+ }
}
diff --git a/jmslib/src/main/java/com/redhat/mqe/lib/JmsMessagingException.java b/lib/src/main/java/com/redhat/mqe/lib/MessagingException.java
similarity index 82%
rename from jmslib/src/main/java/com/redhat/mqe/lib/JmsMessagingException.java
rename to lib/src/main/java/com/redhat/mqe/lib/MessagingException.java
index 203fb40a..7a50907e 100644
--- a/jmslib/src/main/java/com/redhat/mqe/lib/JmsMessagingException.java
+++ b/lib/src/main/java/com/redhat/mqe/lib/MessagingException.java
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017 Red Hat, Inc.
+ * Copyright (c) 2021 Red Hat, Inc.
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -23,19 +23,17 @@
import java.io.ObjectOutputStream;
@SuppressWarnings("serial")
-public class JmsMessagingException extends RuntimeException {
+public class MessagingException extends RuntimeException {
private void writeObject(ObjectOutputStream out)
throws IOException {
}
- public JmsMessagingException(String message) {
+ public MessagingException(String message) {
super(message);
}
- ;
-
- public JmsMessagingException(String message, Throwable cause) {
+ public MessagingException(String message, Throwable cause) {
super(message, cause);
}
diff --git a/jmslib/src/main/java/com/redhat/mqe/lib/Utils.java b/lib/src/main/java/com/redhat/mqe/lib/Utils.java
similarity index 67%
rename from jmslib/src/main/java/com/redhat/mqe/lib/Utils.java
rename to lib/src/main/java/com/redhat/mqe/lib/Utils.java
index 0a75a7e3..61f3a5f9 100644
--- a/jmslib/src/main/java/com/redhat/mqe/lib/Utils.java
+++ b/lib/src/main/java/com/redhat/mqe/lib/Utils.java
@@ -19,13 +19,11 @@
package com.redhat.mqe.lib;
-import org.apache.log4j.Level;
-import org.apache.log4j.LogManager;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.jms.*;
-import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
@@ -39,7 +37,7 @@
*/
public class Utils {
- private static Logger LOG = LoggerFactory.getLogger(Utils.class);
+ private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
public static final List> CLASSES = Collections.unmodifiableList(Arrays.asList(
Integer.class, Long.class, Float.class, Double.class, Boolean.class, String.class));
@@ -77,35 +75,6 @@ public static void sleep(long miliseconds) {
}
}
- /**
- * Calculate TTL of given message from message
- * expiration time and message timestamp.
- *
- * Returns the time the message expires, which is the sum of the time-to-live value
- * specified by the client and the GMT at the time of the send
- * EXP_TIME = CLIENT_SEND+TTL (CLIENT_SEND??)
- * CLIENT_SEND time is approximately getJMSTimestamp() (time value between send()/publish() and return)
- * TODO - check for correctness
- *
- * @param message calculate TTL for this message
- * @return positive long number if TTL was calculated. Long.MIN_VALUE if error.
- */
- public static long getTtl(Message message) {
- long ttl = 0;
- try {
- long expiration = message.getJMSExpiration();
- long timestamp = message.getJMSTimestamp();
- if (expiration != 0 && timestamp != 0) {
- ttl = expiration - timestamp;
- }
- } catch (JMSException jmse) {
- LOG.error("Error while calculating TTL value.\n" + jmse.getMessage());
- jmse.printStackTrace();
- System.exit(1);
- }
- return ttl;
- }
-
/**
* @return number of seconds (including milisecs) since EPOCH.
*/
@@ -127,9 +96,7 @@ public static void sleepUntilNextIteration(double initialTimestamp, int msgCount
if ((duration > 0) && (msgCount > 0)) {
// initial overall duration approximation of whole loop (sender/receiver)
double cummulative_dur = (1.0 * nextCountIndex * duration) / msgCount;
- while (true) {
- if (getTime() - initialTimestamp - cummulative_dur > -0.05)
- break;
+ while (!(getTime() - initialTimestamp - cummulative_dur > -0.05)) {
try {
LOG.trace("sleeping");
Thread.sleep(100);
@@ -145,13 +112,12 @@ public static void sleepUntilNextIteration(double initialTimestamp, int msgCount
* specified, default level is used from *logger* properties file.
* (As of the time writing - simplelogger.properties is used as default.)
*
- * NOTE: SLF4J is not capable of changing log levels programatically!
+ * NOTE: SLF4J is not capable of changing log levels programmatically!
* We have to change the System/File property of given underlying logger.
*
* @param logLevel logging level to be logger set to
*/
public static void setLogLevel(String logLevel) {
- org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger("com.redhat.mqe.jms");
Level level;
switch (logLevel.toLowerCase()) {
case "all":
@@ -181,8 +147,8 @@ public static void setLogLevel(String logLevel) {
default:
level = Level.INFO;
}
- LogManager.getRootLogger().setLevel(level);
- logger.setLevel(level);
+ ((org.apache.logging.log4j.core.Logger) LogManager.getRootLogger()).setLevel(level);
+ ((org.apache.logging.log4j.core.Logger) LogManager.getLogger("com.redhat.mqe.lib")).setLevel(level);
}
/**
@@ -197,7 +163,7 @@ public static void setLogLevel(String logLevel) {
* @return Class type of given value
*/
- public static Class getClassType(String preferredStringType, String value, boolean allowExplicitRetype) throws JmsMessagingException {
+ public static Class getClassType(String preferredStringType, String value, boolean allowExplicitRetype) throws MessagingException {
LOG.trace("getClassType for " + value + ":" + preferredStringType);
if (preferredStringType == null) {
// set manually the contentType to (autotypecasting)
@@ -316,7 +282,7 @@ public static Object getObjectValue(Class clazz, String object, boolean allow
Class argType;
// if it is a subclass of Number
if (Number.class.isAssignableFrom(clazz)
- || object.toLowerCase().equals("true") || object.toLowerCase().equals("false")) {
+ || object.equalsIgnoreCase("true") || object.equalsIgnoreCase("false")) {
argType = String.class;
} else {
argType = Object.class;
@@ -326,75 +292,8 @@ public static Object getObjectValue(Class clazz, String object, boolean allow
return myObj;
}
- public static void streamMessageContentToFile(String filePath, Message message, int msgCounter) {
- try {
- File outputFile = getFilePath(filePath, msgCounter);
- try (FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
- BufferedOutputStream bufferedOutput = new BufferedOutputStream(fileOutputStream)) {
-// final String saveStream = "JMS_AMQ_SaveStream";
- final String saveStream = "JMS_AMQ_OutputStream";
- message.setObjectProperty(saveStream, bufferedOutput);
- }
- } catch (IOException e) {
- LOG.error("Error while writing to file '" + filePath + "'.");
- e.printStackTrace();
- } catch (JMSException e) {
- e.printStackTrace();
- }
- }
-
- /**
- * Write message body (text or binary) to provided file or default one in temp directory.
- *
- * @param filePath file to write data to
- * @param message to be read and written to provided file
- */
- public static void writeMessageContentToFile(String filePath, Message message, int msgCounter) {
- byte[] readByteArray;
- try {
- File file;
- file = getFilePath(filePath, msgCounter);
-
- LOG.debug("Write message content to file '" + file.getPath() + "'.");
- if (message instanceof BytesMessage) {
- LOG.debug("Writing BytesMessage to file");
- BytesMessage bm = (BytesMessage) message;
- readByteArray = new byte[(int) bm.getBodyLength()];
- bm.reset(); // added to be able to read message content
- bm.readBytes(readByteArray);
- try (FileOutputStream fos = new FileOutputStream(file)) {
- fos.write(readByteArray);
- }
- } else if (message instanceof StreamMessage) {
- LOG.debug("Writing StreamMessage to file");
- StreamMessage sm = (StreamMessage) message;
-// sm.reset(); TODO haven't tested this one
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- ObjectOutputStream oos = new ObjectOutputStream(baos);
- oos.writeObject(sm.readObject());
- oos.close();
- } else if (message instanceof TextMessage) {
- LOG.debug("Writing TextMessage to file");
- try (FileWriter fileWriter = new FileWriter(file)) {
- TextMessage tm = (TextMessage) message;
- fileWriter.write(tm.getText());
- }
- }
- } catch (JMSException e) {
- e.printStackTrace();
- } catch (IOException e1) {
- LOG.error("Error while writing to file '" + filePath + "'.");
- e1.printStackTrace();
- }
- }
-
- private static File getFilePath(String filePath, int msgCounter) throws IOException {
- File file;
- if (filePath == null || filePath.equals("")) {
- file = File.createTempFile("recv_msg_", Long.toString(System.currentTimeMillis()));
- } else {
- file = new File(filePath + "_" + msgCounter);
- }
- return file;
+ public static boolean convertOptionToBoolean(String optionStringValue) {
+ String optionValue = optionStringValue.toLowerCase();
+ return (optionValue.equals("true") || optionValue.equals("yes"));
}
}
diff --git a/parent/pom.xml b/parent/pom.xml
index 74a06cb5..cb49fcb6 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -44,7 +44,15 @@
org.slf4j
- slf4j-log4j12
+ slf4j-api
+
+
+ org.apache.logging.log4j
+ log4j-slf4j2-impl
+
+
+ org.apache.logging.log4j
+ log4j-core
net.sf.jopt-simple
@@ -121,6 +129,7 @@
tests
+
@@ -128,30 +137,6 @@
tests
-
-
-
- kotlin-maven-plugin
- org.jetbrains.kotlin
- ${kotlin.version}
-
-
- test-compile
-
- test-compile
-
-
- 1.8
-
- ${project.basedir}/src/test/kotlin
- ${project.basedir}/src/test/java
-
-
-
-
-
-
-
coverage
@@ -197,11 +182,13 @@
${jar.finalName}-${library.version}
+ false
${jar.main.class}
+ true
@@ -234,16 +221,19 @@
${excludeTests}
+
org.apache.maven.plugins
maven-compiler-plugin
${plugin.compiler.version}
- 1.8
- 1.8
+ 11
+ 11
-Xlint:all
true
true
+
+ true
com.google.dagger
@@ -252,22 +242,63 @@
+
+
+
+ default-compile
+ none
+
+
+
+ default-testCompile
+ none
+
+
+ java-compile
+ compile
+
+ compile
+
+
+
+ java-test-compile
+ test-compile
+
+ testCompile
+
+
+
kotlin-maven-plugin
org.jetbrains.kotlin
${kotlin.version}
+
+ compile
+
+ compile
+
+
+ 11
+
+ ${project.basedir}/src/main/kotlin
+ ${project.basedir}/src/main/java
+
+
+
test-compile
test-compile
+ 11
${project.basedir}/src/test/kotlin
${project.basedir}/src/test/java
- ${project.build.directory}/generated-sources/annotations
+
+ ${project.build.directory}/generated-test-sources/transformed
diff --git a/pom.xml b/pom.xml
index f8d642ce..866bd0e6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -29,8 +29,8 @@
pom
- 1.8
- 1.8
+ 11
+ 11
UTF-8
@@ -41,40 +41,21 @@
lib
jmslib
+ jakartalib
cli
cli-activemq
cli-artemis-jms
cli-paho-java
+ cli-protonj2
cli-qpid-jms
+ cli-qpid-jms-1x
+
+ cli-activemq-jmx
+
broker
-
-
-
-
- false
-
- bintray
- https://jcenter.bintray.com
-
-
-
-
-
- false
-
- bintray-plugins
- https://jcenter.bintray.com
-
-
+ interop-tests
+
-
-
- interop-tests
-
- interop-tests
-
-
-
diff --git a/scripts/broker.xml.patch b/scripts/broker.xml.patch
index 94f384e4..e641bb80 100644
--- a/scripts/broker.xml.patch
+++ b/scripts/broker.xml.patch
@@ -14,7 +14,7 @@ index 2ef08a9..4e61d0b 100644
@@ -158,9 +158,11 @@ under the License.
- tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true
+ tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true;supportAdvisory=false;suppressInternalManagementObjects=false
+ tcp://0.0.0.0:61617?sslEnabled=true;keyStorePath=server-side-keystore.jks;keyStorePassword=secureexample;tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true
diff --git a/scripts/entrypoint.sh b/scripts/entrypoint.sh
index d6ff8ce3..157cdc01 100755
--- a/scripts/entrypoint.sh
+++ b/scripts/entrypoint.sh
@@ -32,7 +32,8 @@ if [ ! "$(ls -A /var/lib/amq7/etc)" ]; then
fi
# Log to tty to enable docker logs container-name
-sed -ie "s/logger.handlers=.*/logger.handlers=CONSOLE/g" ../etc/logging.properties
+# TODO: this no longer works with log4j2 logging
+#sed -ie "s/logger.handlers=.*/logger.handlers=CONSOLE/g" ../etc/logging.properties
# Update min memory if the argument is passed
if [[ "$ARTEMIS_MIN_MEMORY" ]]; then
diff --git a/tests.sh b/tests.sh
new file mode 100644
index 00000000..f8452954
--- /dev/null
+++ b/tests.sh
@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+set -x
+
+#
+# Copyright (c) 2021 Red Hat, Inc.
+#
+# 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+#
+
+java -jar cli-activemq-jmx/target/amqx-*.jar --help
+#java -jar cli-activemq-jmx/target/amqx-*.jar queue --host 127.0.0.1:1099 --username admin --password admin --action add --name 'test_default_username_right_password_right'
+
+java -jar cli-activemq/target/cli-activemq-1.2.2-SNAPSHOT-*.jar sender --address cli-activemq --log-msgs json --count 1
+java -jar cli-activemq/target/cli-activemq-1.2.2-SNAPSHOT-*.jar receiver --address cli-activemq --log-msgs json --count 1
+java -jar cli-activemq/target/cli-activemq-1.2.2-SNAPSHOT-*.jar sender --conn-username test --conn-ssl-verify-host false --conn-password test --msg-content 'msg no. %d' --broker ssl://127.0.0.1:61617 --conn-auth-mechanisms PLAIN --timeout 30 --log-msgs json --log-lib trace --address message-basiccli_jms --count 10 --conn-ssl-trust-all true
+
+java -jar cli-artemis-jms/target/cli-artemis-jms-1.2.2-SNAPSHOT-*.jar sender --address cli-artemis-jms --log-msgs json --count 1
+java -jar cli-artemis-jms/target/cli-artemis-jms-1.2.2-SNAPSHOT-*.jar receiver --address cli-artemis-jms --log-msgs json --count 1
+java -jar cli-artemis-jms/target/cli-artemis-jms-1.2.2-SNAPSHOT-*.jar sender --conn-username test --conn-ssl-verify-host false --conn-password test --msg-content 'msg no. %d' --broker tcp://127.0.0.1:61617 --conn-auth-mechanisms PLAIN --timeout 30 --log-msgs json --log-lib trace --address message-basiccli_jms --count 10 --conn-ssl-trust-all true
+
+java -jar cli-paho-java/target/cli-paho-java-1.2.2-SNAPSHOT-*.jar sender --address cli-paho-java --log-msgs json --count 1
+
+cli_qpid_jms_jar=$(find cli-qpid-jms/target -name 'cli-qpid-jms-1.2.2-SNAPSHOT-*.jar' -not -name '*-tests.jar')
+java -jar "${cli_qpid_jms_jar}" sender --address cli-qpid-jms --log-msgs json --count 1
+java -jar "${cli_qpid_jms_jar}" receiver --address cli-qpid-jms --log-msgs json --count 1
+java -jar "${cli_qpid_jms_jar}" sender --conn-username test --conn-ssl-verify-host false --conn-password test --msg-content 'msg no. %d' --broker amqps://127.0.0.1:5673 --conn-auth-mechanisms PLAIN --timeout 30 --log-msgs json --log-lib trace --address message-basiccli_jms --count 10 --conn-ssl-trust-all true
+
+cli_protonj2=$(find cli-protonj2/target -name 'cli-protonj2-1.2.2-SNAPSHOT-*.jar')
+java -jar "${cli_protonj2}" sender --broker amqp://127.0.0.1 --address cli-qpid-jms --log-msgs dict --count 1
+java -jar "${cli_protonj2}" receiver --broker amqp://127.0.0.1 --address cli-qpid-jms --log-msgs dict --count 1
+java -jar "${cli_protonj2}" sender --conn-username test --conn-ssl-verify-peer false --conn-ssl-verify-peer-name false --conn-password test --msg-content 'msg no. %d' --broker amqps://127.0.0.1:5673 --conn-auth-mechanisms PLAIN --timeout 30 --log-msgs json --log-lib trace --address message-basiccli_jms --count 10
diff --git a/tests/pom.xml b/tests/pom.xml
index fe83715c..a07e9afb 100644
--- a/tests/pom.xml
+++ b/tests/pom.xml
@@ -48,10 +48,33 @@
jmslib
+
com.google.truth
truth
- test
+ compile
+
+
+ org.junit-pioneer
+ junit-pioneer
+ compile
+
+
+ org.awaitility
+ awaitility
+ compile
+
+
+ org.apache.activemq
+ artemis-server
+ compile
+
+
+ com.redhat.cli-java
+ broker
+ 1.2.2-SNAPSHOT
+ test-jar
+ compile
diff --git a/tests/src/test/kotlin/AbstractMainTest.kt b/tests/src/test/kotlin/AbstractMainTest.kt
index 7e313848..2cd3a0b0 100644
--- a/tests/src/test/kotlin/AbstractMainTest.kt
+++ b/tests/src/test/kotlin/AbstractMainTest.kt
@@ -17,26 +17,39 @@
* limitations under the License.
*/
+import com.google.common.truth.Truth
import com.google.common.truth.Truth.assertThat
+import com.google.common.truth.TruthJUnit
import com.redhat.mqe.ClientListener
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings
+import org.awaitility.Awaitility.await
import org.junit.jupiter.api.Assertions.assertTimeoutPreemptively
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Tag
import org.junit.jupiter.api.Tags
import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
import org.junit.jupiter.api.function.Executable
+import org.junit.jupiter.api.io.TempDir
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.CsvFileSource
import org.junit.jupiter.params.provider.ValueSource
+import util.Broker
+import util.BrokerFixture
import java.io.File
+import java.lang.reflect.UndeclaredThrowableException
import java.math.BigInteger
import java.nio.file.Files
+import java.nio.file.Path
import java.security.MessageDigest
-import java.security.Permission
import java.time.Duration
+import java.time.Instant
import java.time.LocalTime
-import kotlin.collections.ArrayList
-import kotlin.test.fail
+import java.util.*
+import java.util.concurrent.Callable
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
@Tag("external")
abstract class AbstractMainTest : AbstractTest() {
@@ -56,6 +69,11 @@ abstract class AbstractMainTest : AbstractTest() {
abstract fun main_(listener: ClientListener, args: Array)
fun main(args: Array): List> {
val messages = ArrayList>()
+ main(args, messages)
+ return messages
+ }
+
+ fun main(args: Array, messages: MutableList>): List> {
main_(object : ClientListener {
override fun onMessage(message: Map) {
messages.add(message)
@@ -80,21 +98,13 @@ abstract class AbstractMainTest : AbstractTest() {
@Tags(Tag("pairwise"), Tag("external"))
@ParameterizedTest
- @ValueSource(strings = arrayOf("sender", "receiver", "connector"))
+ @ValueSource(strings = ["sender", "receiver", "connector"])
fun printHelp(client: String) {
val parameters =
"$client --help".split(" ").toTypedArray()
- val previousManager = System.getSecurityManager()
- try {
- val manager = NoExitSecurityManager(previousManager)
- System.setSecurityManager(manager)
+ assertSystemExit(0, Executable {
main(parameters)
- fail("expected exception")
- } catch (e: SystemExitingWithStatus) {
- assertThat(e.status).isEqualTo(0)
- } finally {
- System.setSecurityManager(previousManager)
- }
+ })
}
@Tag("external")
@@ -136,7 +146,9 @@ abstract class AbstractMainTest : AbstractTest() {
"connector --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
print("Connecting: ")
- main(connectorParameters)
+ assertNoSystemExit {
+ main(connectorParameters)
+ }
}
}
@@ -144,9 +156,13 @@ abstract class AbstractMainTest : AbstractTest() {
@Test
fun sendAndReceiveSingleMessageUsingCredentials() {
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --conn-username admin --conn-password admin --count 1".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --conn-username admin --conn-password admin --count 1".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --broker $brokerUrl --address $address --conn-username admin --conn-password admin --count 1".split(" ").toTypedArray()
+ "receiver --log-msgs dict --broker $brokerUrl --address $address --conn-username admin --conn-password admin --count 1".split(
+ " "
+ ).toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
print("Sending: ")
main(senderParameters)
@@ -155,13 +171,29 @@ abstract class AbstractMainTest : AbstractTest() {
}
}
+ @Tag("external")
+ @Test
+ fun test_simple_transaction_sender_batch_size_leftovers() {
+ print("Sending:\n ")
+ val sent = main(
+ "sender --log-msgs interop --broker $brokerUrl --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin --address $address --count 11 --tx-size 3 --tx-action commit".split(" ").toTypedArray()
+ )
+ print("Receiving:\n ")
+ val received = main(
+ "receiver --timeout 10 --log-msgs interop --broker $brokerUrl --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin --address $address --count 0".split(" ").toTypedArray()
+ )
+ assertThat(sent).hasSize(11)
+ assertThat(received).hasSize(9)
+ }
+
@Tag("external")
@Test
fun sendBrowseAndReceiveSingleMessageWithEmptySelector() {
val senderParameters =
"sender --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --broker $brokerUrl --address $address --msg-selector '' --count 1".split(" ").toTypedArray()
+ "receiver --log-msgs dict --broker $brokerUrl --address $address --msg-selector= --count 1".split(" ")
+ .toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
print("Sending: ")
main(senderParameters)
@@ -172,10 +204,41 @@ abstract class AbstractMainTest : AbstractTest() {
}
}
+ /**
+ * Sends two messages and then uses selector to receive the second message.
+ *
+ * If selector was (erroneously) not applied, the first message would be received here.
+ */
+ @Tag("external")
+ @Test
+ fun sendBrowseAndReceiveSingleMessageWithNonemptySelector() {
+ val senderParameters1 =
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-property=a~1".split(" ").toTypedArray()
+ val senderParameters2 =
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
+ val receiverParameters =
+ "receiver / --log-msgs=dict / --broker=$brokerUrl / --address=$address / --msg-selector=a IS NULL / --timeout=2".split(" / ").toTypedArray()
+ assertTimeoutPreemptively(Duration.ofSeconds(10)) {
+ print("Sending: ")
+ main(senderParameters1)
+ main(senderParameters2)
+ print("Browsing: ")
+ val browsed = main(receiverParameters + "--recv-browse true".split(" ").toTypedArray())
+ assertThat(browsed).hasSize(1)
+ // can't assert empty properties, openwire adds its own,
+ // `{__AMQ_CID=ID:fedora.jiridanek-42217-1666942674180-2:1}`
+ assertThat(browsed[0]["properties"] as Map).doesNotContainKey("a")
+ print("Receiving: ")
+ val received = main(receiverParameters)
+ assertThat(received).hasSize(1)
+ assertThat(received[0]["properties"] as Map).doesNotContainKey("a")
+ }
+ }
+
@Tag("external")
@Test
fun sendSingleMessageWithoutProtocolInBrokerUrl() {
- val brokerUrl = brokerUrl.substringAfter(":")
+ val brokerUrl = brokerUrl.substringAfterLast("/")
val senderParameters =
"sender --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
@@ -203,9 +266,11 @@ abstract class AbstractMainTest : AbstractTest() {
@Test
fun sendAndReceiveSingleMessageLogJson() {
val senderParameters =
- "sender --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
+ "sender --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ")
+ .toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
+ "receiver --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ")
+ .toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
print("Sending: ")
main(senderParameters)
@@ -218,9 +283,12 @@ abstract class AbstractMainTest : AbstractTest() {
@Test
fun sendAndReceiveSingleMessageLogJsonFloatType() {
val senderParameters =
- "sender --log-msgs dict --out json --broker $brokerUrl --address $address --msg-property baf~42.2 --count 1".split(" ").toTypedArray()
+ "sender --log-msgs dict --out json --broker $brokerUrl --address $address --msg-property baf~42.2 --count 1".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
+ "receiver --log-msgs dict --out json --broker $brokerUrl --address $address --count 1".split(" ")
+ .toTypedArray()
assertTimeoutPreemptively(Duration.ofSeconds(10)) {
print("Sending: ")
main(senderParameters)
@@ -231,8 +299,8 @@ abstract class AbstractMainTest : AbstractTest() {
@Tags(Tag("pairwise"), Tag("external"))
@ParameterizedTest
- @CsvFileSource(resources = arrayOf("/receiver.csv"))
- fun sendAndReceiveWithAllReceiverCLISwitches(receiverDynamicOptions: String) {
+ @CsvFileSource(resources = ["/receiver.csv"])
+ open fun sendAndReceiveWithAllReceiverCLISwitches(receiverDynamicOptions: String) {
println(receiverDynamicOptions)
val senderParameters =
"sender --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
@@ -249,8 +317,8 @@ abstract class AbstractMainTest : AbstractTest() {
@Tags(Tag("pairwise"), Tag("external"))
@ParameterizedTest
- @CsvFileSource(resources = arrayOf("/sender.csv"))
- fun sendAndReceiveWithAllSenderCLISwitches(senderDynamicOptions: String) {
+ @CsvFileSource(resources = ["/sender.csv"])
+ open fun sendAndReceiveWithAllSenderCLISwitches(senderDynamicOptions: String) {
println(senderDynamicOptions)
val senderParameters =
"sender --broker $brokerUrl --address $address".split(" ").toTypedArray()
@@ -267,8 +335,8 @@ abstract class AbstractMainTest : AbstractTest() {
@Tags(Tag("pairwise"), Tag("external"))
@ParameterizedTest
- @CsvFileSource(resources = arrayOf("/connector.csv"))
- fun connectConnectorWithAllSenderCLISwitches(senderDynamicOptions: String) {
+ @CsvFileSource(resources = ["/connector.csv"])
+ open fun connectConnectorWithAllSenderCLISwitches(senderDynamicOptions: String) {
println(senderDynamicOptions)
val connectorPrameters =
"connector --broker $brokerUrl --address $address".split(" ").toTypedArray()
@@ -283,7 +351,9 @@ abstract class AbstractMainTest : AbstractTest() {
@Test
fun sendAndReceiveListMessage() {
val senderParameters =
- """sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-list-item --msg-content-list-item "String" --msg-content-list-item "~1" --msg-content-list-item "~1.0" --msg-content-list-item "1" --msg-content-list-item "1.0" --msg-content-list-item "~-1" --msg-content-list-item "~-1.3" --msg-content-list-item "-1" --msg-content-list-item "~~1"""".split(" ").toTypedArray()
+ """sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-list-item --msg-content-list-item "String" --msg-content-list-item "~1" --msg-content-list-item "~1.0" --msg-content-list-item "1" --msg-content-list-item "1.0" --msg-content-list-item "~-1" --msg-content-list-item "~-1.3" --msg-content-list-item "-1" --msg-content-list-item "~~1"""".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
"receiver --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
@@ -298,7 +368,9 @@ abstract class AbstractMainTest : AbstractTest() {
fun sendAndReceiveAnInt() {
assertNoSystemExit {
val senderParameters =
- """sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --content-type int --msg-content 1234""".split(" ").toTypedArray()
+ """sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --content-type int --msg-content 1234""".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
"receiver --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
@@ -316,7 +388,9 @@ abstract class AbstractMainTest : AbstractTest() {
try {
file.writeText("aContent")
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
"receiver --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
@@ -336,7 +410,9 @@ abstract class AbstractMainTest : AbstractTest() {
try {
file.writeText("aContent")
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file --msg-content-binary true".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file --msg-content-binary true".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
"receiver --log-msgs dict --broker $brokerUrl --address $address --count 1".split(" ").toTypedArray()
@@ -355,7 +431,9 @@ abstract class AbstractMainTest : AbstractTest() {
assertSystemExit(2, Executable {
val file = "noSuchFile"
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(
+ " "
+ ).toTypedArray()
main(senderParameters)
})
}
@@ -366,7 +444,9 @@ abstract class AbstractMainTest : AbstractTest() {
assertSystemExit(2, Executable {
val file = "noSuchFile"
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file --msg-content-binary true".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file --msg-content-binary true".split(
+ " "
+ ).toTypedArray()
main(senderParameters)
})
}
@@ -381,9 +461,13 @@ abstract class AbstractMainTest : AbstractTest() {
try {
file.writeText("aContent")
val senderParameters =
- "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-from-file $file".split(
+ " "
+ ).toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-to-file $output".split(" ").toTypedArray()
+ "receiver --log-msgs dict --broker $brokerUrl --address $address --count 1 --msg-content-to-file $output".split(
+ " "
+ ).toTypedArray()
print("Sending: ")
main(senderParameters)
@@ -403,7 +487,8 @@ abstract class AbstractMainTest : AbstractTest() {
val senderParameters =
"sender --log-msgs dict --broker $brokerUrl --address topic://$address --count 1".split(" ").toTypedArray()
val receiverParameters =
- "receiver --log-msgs dict --broker $brokerUrl --address topic://$address --count 1".split(" ").toTypedArray()
+ "receiver --log-msgs dict --broker $brokerUrl --address topic://$address --count 1".split(" ")
+ .toTypedArray()
val t = Thread {
print("Receiving: ")
@@ -422,7 +507,9 @@ abstract class AbstractMainTest : AbstractTest() {
assertNoSystemExit {
// "sender --log-msgs dict --broker tcp://127.0.0.1:61617 --address lalaLand --count 1 --conn-ssl-truststore-location --conn-ssl-truststore-password secureexample --conn-ssl-keystore-location /home/jdanek/Downloads/AMQ7/amq7cr1i0/ikeysiold/client-side-keystore.jks --conn-ssl-keystore-password secureexample".split(" ").toTypedArray()
val senderParameters =
- "sender --log-msgs dict --broker $sslBrokerUrl --address $address --conn-ssl-verify-host false --conn-ssl-trust-all true --count 1".split(" ").toTypedArray()
+ "sender --log-msgs dict --broker $sslBrokerUrl --address $address --conn-ssl-verify-host false --conn-ssl-trust-all true --count 1".split(
+ " "
+ ).toTypedArray()
print("Sending: ")
main(senderParameters)
}
@@ -449,13 +536,235 @@ abstract class AbstractMainTest : AbstractTest() {
val expected = BigInteger(1, md.digest(content.toByteArray())).toString(16)
print(expected)
assertNoSystemExit {
- val sent = main(arrayOf(
- "sender", "--log-msgs", "dict", "--broker", brokerUrl, "--address", address, "--count", "1", "--msg-content", content, "--msg-content-hashed"))
- val received = main(arrayOf(
- "receiver", "--log-msgs", "dict", "--broker", brokerUrl, "--address", address, "--count", "1", "--msg-content-hashed"))
+ val sent = main(
+ arrayOf(
+ "sender",
+ "--log-msgs",
+ "dict",
+ "--broker",
+ brokerUrl,
+ "--address",
+ address,
+ "--count",
+ "1",
+ "--msg-content",
+ content,
+ "--msg-content-hashed"
+ )
+ )
+ val received = main(
+ arrayOf(
+ "receiver",
+ "--log-msgs",
+ "dict",
+ "--broker",
+ brokerUrl,
+ "--address",
+ address,
+ "--count",
+ "1",
+ "--msg-content-hashed"
+ )
+ )
assertThat(sent.map { it["content"] as String }).containsExactlyElementsIn(listOf(expected))
assertThat(received.map { it["content"] as String }).containsExactlyElementsIn(listOf(expected))
}
}
+
+ @Tag("external")
+ @Test
+ fun testSendingReceivingWithDuration() {
+ assertNoSystemExit {
+ val senderParameters =
+ "sender --log-msgs dict --broker $brokerUrl --address $address --count=5 --duration=5".split(" ").toTypedArray()
+ val receiverParameters =
+ "receiver --log-msgs dict --broker $brokerUrl --address $address --count=5 --duration=5".split(" ").toTypedArray()
+ print("Sending:\n ")
+ main(senderParameters)
+ print("Receiving:\n ")
+ main(receiverParameters)
+ }
+ }
+
+ // void testMessageContentListItem() {
+ // '--msg-content-list-item', '', '--msg-content-list-item', 'String', '--msg-content-list-item', '~1', '--msg-content-list-item', '~1.0', '--msg-content-list-item', '1', '--msg-content-list-item', '1.0', '--msg-content-list-item', '~-1', '--msg-content-list-item', '~-1.3', '--msg-content-list-item', '-1', '--msg-content-list-item', '~~1'
+ // }
+
+ @Tag("external")
+ // @SetEnvironmentVariable(key = "PN_TRACE_FRM", value = "true")
+ @Test
+ @Throws(
+ Throwable::class
+ )
+ open fun testDurableSubscriber() {
+
+ // tests.JAMQNode000Tests.JAMQNodeTests.test_node_durable_topic_subscriber
+ // topic: prefix, --conn-clientid --durable-subscriber --durable-subscriber-name
+ println()
+ println("Create subscriber")
+ println()
+ main(
+ ("receiver --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 0 --durable-subscriber True" +
+ " --conn-clientid cliId0 --durable-subscriber-name ds0").split(" ").toTypedArray()
+ )
+
+ println()
+ println("Send message")
+ println()
+
+ main(
+ ("sender --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 1").split(" ").toTypedArray()
+ )
+
+ println()
+ println("Recover subscription, receive message")
+ println()
+
+ main(
+ ("receiver --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 0 --durable-subscriber True" +
+ " --conn-clientid cliId0 --durable-subscriber-name ds0").split(" ").toTypedArray()
+ )
+
+ println()
+ println("Unsubscribe")
+ println()
+
+ // --subscriber-unsubscribe True
+ main(("receiver --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 0" +
+ " --subscriber-unsubscribe True" +
+ " --conn-clientid cliId0 --durable-subscriber-name ds0").split(" ").toTypedArray()
+ )
+
+ println()
+ println("Send message again")
+ println()
+
+ main(
+ ("sender --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 1").split(" ").toTypedArray()
+ )
+
+ println()
+ println("Try receiving, message should not be there")
+ println()
+
+ val rcv2 = main(
+ ("receiver --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 0 --durable-subscriber True" +
+ " --conn-clientid cliId0 --durable-subscriber-name ds0").split(" ").toTypedArray()
+ )
+ assertThat(rcv2).hasSize(0)
+
+ println()
+ println("Unsubscribe again")
+ println()
+
+ // --subscriber-unsubscribe True
+ main(("receiver --timeout 5 --log-msgs dict --broker " + brokerUrl + " --conn-auth-mechanisms PLAIN --conn-username admin --conn-password admin" +
+ " --address topic://test_node_durable_topic_subscriber --count 0" +
+ " --subscriber-unsubscribe True" +
+ " --conn-clientid cliId0 --durable-subscriber-name ds0").split(" ").toTypedArray()
+ )
+ }
+
+ /**
+ * Sends sufficient amount of messages to cause broker to block the sender.
+ * Then it runs a receiver and empties the queue, unblocking the queue, and receiving all messages eventually.
+ *
+ * See dTests test JAMQNode000Tests/test_broker_blocks_client_many_messages_in_queue
+ */
+ @Test
+ @ExtendWith(BrokerFixture::class)
+ fun testQueueBlockUnblockSenderResumes(@BrokerFixture.TempBroker broker: Broker, @TempDir tempDir: Path) {
+ // this does not work with qpid-jms, and we skip the test in dtests anyways
+ TruthJUnit.assume().that(this::class.java.name).endsWith("ProtonJ2MainTest")
+
+ val address = "test_broker_blocks_client_many_messages_in_queue"
+
+ broker.configuration.isSecurityEnabled = false
+ broker.configuration.bindingsDirectory = tempDir.resolve("data/bindings").toString()
+ broker.configuration.journalDirectory = tempDir.resolve("data/journal").toString()
+ broker.configuration.largeMessagesDirectory = tempDir.resolve("data/large").toString()
+ broker.configuration.pagingDirectory = tempDir.resolve("data/paging").toString()
+ val addressSettings = AddressSettings()
+ .setMaxSizeBytes(10240)
+ .setMaxSizeBytesRejectThreshold(10240)
+ .setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK)
+ broker.configuration.addAddressSetting(address, addressSettings)
+ broker.startBroker()
+
+ val brokerUrl = "localhost:" + broker.addAMQPAcceptor()
+
+ val pool = Executors.newCachedThreadPool()
+
+ val senderMessages = Collections.synchronizedList>(java.util.ArrayList())
+ val senderFuture = pool.submit(Callable>> {
+ main(
+ arrayOf(
+ "sender",
+ "--timeout=60",
+ "--log-msgs=dict",
+ "--msg-content-hashed=true",
+ "--broker=$brokerUrl",
+ "--conn-auth-mechanisms=PLAIN",
+ "--conn-username=admin",
+ "--conn-password=admin",
+ "--address=$address",
+ "--count=300",
+ "--msg-content=abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcd"
+ ),
+ senderMessages
+ )
+ })
+
+ val addressControl = broker.makeAddressControl(address)
+ var previousCount: Long = -1
+ await().atMost(5, TimeUnit.SECONDS).ignoreException(UndeclaredThrowableException::class.java).until {
+ previousCount = addressControl.messageCount
+ true
+ }
+
+ val initialTime = Instant.now()
+
+ var gotBlocked = false
+ while (Duration.between(initialTime, Instant.now()).toSeconds() < 10) {
+ TimeUnit.SECONDS.sleep(2)
+ if (addressControl.messageCount > 0
+ && addressControl.messageCount == previousCount
+ && addressControl.messageCount < 300
+ ) {
+ gotBlocked = true
+ break
+ }
+ previousCount = addressControl.messageCount
+ }
+ Truth.assertWithMessage("Sender should get blocked, $initialTime, ${Instant.now()}, $previousCount, ${senderMessages.size}")
+ .that(gotBlocked).isTrue()
+ println("Sender got blocked, $initialTime, ${Instant.now()}, $previousCount, ${senderMessages.size}")
+
+ val receivedMessages = main(
+ arrayOf(
+ "receiver",
+ "--timeout=60",
+ "--log-msgs=dict",
+ "--msg-content-hashed=true",
+ "--broker=$brokerUrl",
+ "--conn-auth-mechanisms=PLAIN",
+ "--conn-username=admin",
+ "--conn-password=admin",
+ "--address=$address",
+ "--count=300",
+ )
+ )
+
+ assertThat(receivedMessages).hasSize(300)
+
+ senderFuture.get()
+ assertThat(senderMessages).hasSize(300)
+ }
}
diff --git a/tests/src/test/kotlin/testing.kt b/tests/src/test/kotlin/testing.kt
index 81eda005..81fa812c 100644
--- a/tests/src/test/kotlin/testing.kt
+++ b/tests/src/test/kotlin/testing.kt
@@ -34,9 +34,16 @@ fun assertSystemExit(status: Int, executable: Executable) {
val manager = NoExitSecurityManager(previousManager)
System.setSecurityManager(manager)
- executable.execute()
+ try {
+ executable.execute()
+ } catch (t: Throwable) {
+ return // also allow any exception be thrown
+ }
- fail("expected exception")
+ // allow not exitting explicitly (for protonj2)
+ if (status != 0) {
+ fail("expected exception")
+ }
} catch (e: SystemExitingWithStatus) {
Truth.assertThat(e.status).isEqualTo(status)
} finally {
@@ -53,7 +60,7 @@ fun assertNoSystemExit(executable: () -> Unit) {
executable()
} catch (e: SystemExitingWithStatus) {
- fail("System.exit has been called")
+ fail("System.exit has been called", e)
} finally {
System.setSecurityManager(previousManager)
}