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 2021/07/16 13:19:14 UTC
[activemq-artemis] 01/03: NO-JIRA Adding Core Bridge Transfer Test
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 277aa3706a232c8fab62b9144bc0b15120b1634f
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Jul 13 17:24:38 2021 -0400
NO-JIRA Adding Core Bridge Transfer Test
---
tests/smoke-tests/pom.xml | 32 +++
.../servers/bridgeTransfer/serverA/broker.xml | 247 +++++++++++++++++++++
.../servers/bridgeTransfer/serverB/broker.xml | 229 +++++++++++++++++++
.../bridgeTransfer/BridgeTransferingTest.java | 190 ++++++++++++++++
4 files changed, 698 insertions(+)
diff --git a/tests/smoke-tests/pom.xml b/tests/smoke-tests/pom.xml
index 7fb05df..59b7f17 100644
--- a/tests/smoke-tests/pom.xml
+++ b/tests/smoke-tests/pom.xml
@@ -848,6 +848,38 @@
<configuration>${basedir}/target/classes/servers/brokerConnect/qdr</configuration>
</configuration>
</execution>
+ <execution>
+ <phase>test-compile</phase>
+ <id>create-bridge-transfer-A</id>
+ <goals>
+ <goal>create</goal>
+ </goals>
+ <configuration>
+ <role>amq</role>
+ <user>artemis</user>
+ <password>artemis</password>
+ <allowAnonymous>true</allowAnonymous>
+ <noWeb>true</noWeb>
+ <instance>${basedir}/target/bridgeTransfer/serverA</instance>
+ <configuration>${basedir}/target/classes/servers/bridgeTransfer/serverA</configuration>
+ </configuration>
+ </execution>
+ <execution>
+ <phase>test-compile</phase>
+ <id>create-bridge-transfer-B</id>
+ <goals>
+ <goal>create</goal>
+ </goals>
+ <configuration>
+ <role>amq</role>
+ <user>artemis</user>
+ <password>artemis</password>
+ <allowAnonymous>true</allowAnonymous>
+ <noWeb>true</noWeb>
+ <instance>${basedir}/target/bridgeTransfer/serverB</instance>
+ <configuration>${basedir}/target/classes/servers/bridgeTransfer/serverB</configuration>
+ </configuration>
+ </execution>
</executions>
<dependencies>
<dependency>
diff --git a/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverA/broker.xml b/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverA/broker.xml
new file mode 100644
index 0000000..b43b5b5
--- /dev/null
+++ b/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverA/broker.xml
@@ -0,0 +1,247 @@
+<?xml version='1.0'?>
+<!--
+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.
+-->
+
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns:xi="http://www.w3.org/2001/XInclude"
+ xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
+
+ <core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq:core ">
+
+ <name>0.0.0.0</name>
+
+
+ <persistence-enabled>true</persistence-enabled>
+
+ <!-- this could be ASYNCIO, MAPPED, NIO
+ ASYNCIO: Linux Libaio
+ MAPPED: mmap files
+ NIO: Plain Java Files
+ -->
+ <journal-type>NIO</journal-type>
+
+ <paging-directory>./data/paging</paging-directory>
+
+ <bindings-directory>./data/bindings</bindings-directory>
+
+ <journal-directory>./data/journal</journal-directory>
+
+ <large-messages-directory>./data/large-messages</large-messages-directory>
+
+ <!--
+ if you want to retain your journal uncomment this following configuration.
+
+ This will allow your system to keep 7 days of your data, up to 10G. Tweak it accordingly to your use case and capacity.
+
+ it is recommended to use a separate storage unit from the journal for performance considerations.
+
+ <journal-retention-directory period="7" unit="DAYS" storage-limit="10G">data/retention</journal-retention-directory>>
+ -->
+
+ <journal-datasync>true</journal-datasync>
+
+ <journal-min-files>2</journal-min-files>
+
+ <journal-pool-files>10</journal-pool-files>
+
+ <journal-device-block-size>4096</journal-device-block-size>
+
+ <journal-file-size>10M</journal-file-size>
+ <!--
+ You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
+ <network-check-NIC>theNicName</network-check-NIC>
+ -->
+
+ <!--
+ Use this to use an HTTP server to validate the network
+ <network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
+
+ <!-- <network-check-period>10000</network-check-period> -->
+ <!-- <network-check-timeout>1000</network-check-timeout> -->
+
+ <!-- this is a comma separated list, no spaces, just DNS or IPs
+ it should accept IPV6
+
+ Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
+ Using IPs that could eventually disappear or be partially visible may defeat the purpose.
+ You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
+ <!-- <network-check-list>10.0.0.1</network-check-list> -->
+
+ <!-- use this to customize the ping used for ipv4 addresses -->
+ <!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
+
+ <!-- use this to customize the ping used for ipv6 addresses -->
+ <!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
+
+
+
+
+ <!-- how often we are looking for how many bytes are being used on the disk in ms -->
+ <disk-scan-period>5000</disk-scan-period>
+
+ <!-- once the disk hits this limit the system will block, or close the connection in certain protocols
+ that won't support flow control. -->
+ <max-disk-usage>90</max-disk-usage>
+
+ <!-- should the broker detect dead locks and other issues -->
+ <critical-analyzer>true</critical-analyzer>
+
+ <critical-analyzer-timeout>120000</critical-analyzer-timeout>
+
+ <critical-analyzer-check-period>60000</critical-analyzer-check-period>
+
+ <critical-analyzer-policy>HALT</critical-analyzer-policy>
+
+
+
+ <!-- the system will enter into page mode once you hit this limit.
+ This is an estimate in bytes of how much the messages are using in memory
+
+ The system will use half of the available memory (-Xmx) by default for the global-max-size.
+ You may specify a different value here if you need to customize it to your needs.
+
+ <global-max-size>100Mb</global-max-size>
+
+ -->
+
+ <acceptors>
+
+ <!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
+ <!-- amqpCredits: The number of credits sent to AMQP producers -->
+ <!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
+ <!-- amqpDuplicateDetection: If you are not using duplicate detection, set this to false
+ as duplicate detection requires applicationProperties to be parsed on the server. -->
+ <!-- amqpMinLargeMessageSize: Determines how many bytes are considered large, so we start using files to hold their data.
+ default: 102400, -1 would mean to disable large mesasge control -->
+
+ <!-- Note: If an acceptor needs to be compatible with HornetQ and/or Artemis 1.x clients add
+ "anycastPrefix=jms.queue.;multicastPrefix=jms.topic." to the acceptor url.
+ See https://issues.apache.org/jira/browse/ARTEMIS-1644 for more information. -->
+
+
+ <!-- Acceptor for every supported protocol -->
+ <acceptor name="artemis">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</acceptor>
+
+ </acceptors>
+
+ <connectors>
+ <connector name="other-side">tcp://localhost:61617</connector>
+ </connectors>
+
+
+ <bridges>
+ <bridge name="otherSide">
+ <queue-name>bridgeQueue</queue-name>
+ <retry-interval>100</retry-interval>
+ <reconnect-attempts>-1</reconnect-attempts>
+ <static-connectors>
+ <connector-ref>other-side</connector-ref>
+ </static-connectors>
+ </bridge>
+ </bridges>
+
+
+
+
+ <security-settings>
+ <security-setting match="#">
+ <permission type="createNonDurableQueue" roles="amq"/>
+ <permission type="deleteNonDurableQueue" roles="amq"/>
+ <permission type="createDurableQueue" roles="amq"/>
+ <permission type="deleteDurableQueue" roles="amq"/>
+ <permission type="createAddress" roles="amq"/>
+ <permission type="deleteAddress" roles="amq"/>
+ <permission type="consume" roles="amq"/>
+ <permission type="browse" roles="amq"/>
+ <permission type="send" roles="amq"/>
+ <!-- we need this otherwise ./artemis data imp wouldn't work -->
+ <permission type="manage" roles="amq"/>
+ </security-setting>
+ </security-settings>
+
+ <address-settings>
+ <!-- if you define auto-create on certain queues, management has to be auto-create -->
+ <address-setting match="activemq.management#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <!-- with -1 only the global-max-size is in use for limiting -->
+ <max-size-bytes>-1</max-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ <auto-create-jms-queues>true</auto-create-jms-queues>
+ <auto-create-jms-topics>true</auto-create-jms-topics>
+ </address-setting>
+ <!--default for catch all-->
+ <address-setting match="#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <!-- with -1 only the global-max-size is in use for limiting -->
+ <max-size-bytes>-1</max-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ <auto-create-jms-queues>true</auto-create-jms-queues>
+ <auto-create-jms-topics>true</auto-create-jms-topics>
+ </address-setting>
+ </address-settings>
+
+ <addresses>
+ <address name="DLQ">
+ <anycast>
+ <queue name="DLQ" />
+ </anycast>
+ </address>
+ <address name="ExpiryQueue">
+ <anycast>
+ <queue name="ExpiryQueue" />
+ </anycast>
+ </address>
+
+ <address name="bridgeQueue">
+ <anycast>
+ <queue name="bridgeQueue"/>
+ </anycast>
+ </address>
+
+ </addresses>
+
+
+ <!-- Uncomment the following if you want to use the Standard LoggingActiveMQServerPlugin pluging to log in events
+ <broker-plugins>
+ <broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
+ <property key="LOG_ALL_EVENTS" value="true"/>
+ <property key="LOG_CONNECTION_EVENTS" value="true"/>
+ <property key="LOG_SESSION_EVENTS" value="true"/>
+ <property key="LOG_CONSUMER_EVENTS" value="true"/>
+ <property key="LOG_DELIVERING_EVENTS" value="true"/>
+ <property key="LOG_SENDING_EVENTS" value="true"/>
+ <property key="LOG_INTERNAL_EVENTS" value="true"/>
+ </broker-plugin>
+ </broker-plugins>
+ -->
+
+ </core>
+</configuration>
diff --git a/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverB/broker.xml b/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverB/broker.xml
new file mode 100644
index 0000000..8920f41
--- /dev/null
+++ b/tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverB/broker.xml
@@ -0,0 +1,229 @@
+<?xml version='1.0'?>
+<!--
+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.
+-->
+
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns:xi="http://www.w3.org/2001/XInclude"
+ xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
+
+ <core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq:core ">
+
+ <name>0.0.0.0</name>
+
+
+ <persistence-enabled>true</persistence-enabled>
+
+ <!-- this could be ASYNCIO, MAPPED, NIO
+ ASYNCIO: Linux Libaio
+ MAPPED: mmap files
+ NIO: Plain Java Files
+ -->
+ <journal-type>NIO</journal-type>
+
+ <paging-directory>./data/paging</paging-directory>
+
+ <bindings-directory>./data/bindings</bindings-directory>
+
+ <journal-directory>./data/journal</journal-directory>
+
+ <large-messages-directory>./data/large-messages</large-messages-directory>
+
+ <!--
+ if you want to retain your journal uncomment this following configuration.
+
+ This will allow your system to keep 7 days of your data, up to 10G. Tweak it accordingly to your use case and capacity.
+
+ it is recommended to use a separate storage unit from the journal for performance considerations.
+
+ <journal-retention-directory period="7" unit="DAYS" storage-limit="10G">data/retention</journal-retention-directory>>
+ -->
+
+ <journal-datasync>true</journal-datasync>
+
+ <journal-min-files>2</journal-min-files>
+
+ <journal-pool-files>10</journal-pool-files>
+
+ <journal-device-block-size>4096</journal-device-block-size>
+
+ <journal-file-size>10M</journal-file-size>
+ <!--
+ You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
+ <network-check-NIC>theNicName</network-check-NIC>
+ -->
+
+ <!--
+ Use this to use an HTTP server to validate the network
+ <network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
+
+ <!-- <network-check-period>10000</network-check-period> -->
+ <!-- <network-check-timeout>1000</network-check-timeout> -->
+
+ <!-- this is a comma separated list, no spaces, just DNS or IPs
+ it should accept IPV6
+
+ Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
+ Using IPs that could eventually disappear or be partially visible may defeat the purpose.
+ You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
+ <!-- <network-check-list>10.0.0.1</network-check-list> -->
+
+ <!-- use this to customize the ping used for ipv4 addresses -->
+ <!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
+
+ <!-- use this to customize the ping used for ipv6 addresses -->
+ <!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
+
+
+
+
+ <!-- how often we are looking for how many bytes are being used on the disk in ms -->
+ <disk-scan-period>5000</disk-scan-period>
+
+ <!-- once the disk hits this limit the system will block, or close the connection in certain protocols
+ that won't support flow control. -->
+ <max-disk-usage>90</max-disk-usage>
+
+ <!-- should the broker detect dead locks and other issues -->
+ <critical-analyzer>true</critical-analyzer>
+
+ <critical-analyzer-timeout>120000</critical-analyzer-timeout>
+
+ <critical-analyzer-check-period>60000</critical-analyzer-check-period>
+
+ <critical-analyzer-policy>HALT</critical-analyzer-policy>
+
+
+
+ <!-- the system will enter into page mode once you hit this limit.
+ This is an estimate in bytes of how much the messages are using in memory
+
+ The system will use half of the available memory (-Xmx) by default for the global-max-size.
+ You may specify a different value here if you need to customize it to your needs.
+
+ <global-max-size>100Mb</global-max-size>
+
+ -->
+
+ <acceptors>
+
+ <!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
+ <!-- amqpCredits: The number of credits sent to AMQP producers -->
+ <!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
+ <!-- amqpDuplicateDetection: If you are not using duplicate detection, set this to false
+ as duplicate detection requires applicationProperties to be parsed on the server. -->
+ <!-- amqpMinLargeMessageSize: Determines how many bytes are considered large, so we start using files to hold their data.
+ default: 102400, -1 would mean to disable large mesasge control -->
+
+ <!-- Note: If an acceptor needs to be compatible with HornetQ and/or Artemis 1.x clients add
+ "anycastPrefix=jms.queue.;multicastPrefix=jms.topic." to the acceptor url.
+ See https://issues.apache.org/jira/browse/ARTEMIS-1644 for more information. -->
+
+
+ <!-- Acceptor for every supported protocol -->
+ <acceptor name="artemis">tcp://0.0.0.0:61617?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true</acceptor>
+
+ </acceptors>
+
+
+ <security-settings>
+ <security-setting match="#">
+ <permission type="createNonDurableQueue" roles="amq"/>
+ <permission type="deleteNonDurableQueue" roles="amq"/>
+ <permission type="createDurableQueue" roles="amq"/>
+ <permission type="deleteDurableQueue" roles="amq"/>
+ <permission type="createAddress" roles="amq"/>
+ <permission type="deleteAddress" roles="amq"/>
+ <permission type="consume" roles="amq"/>
+ <permission type="browse" roles="amq"/>
+ <permission type="send" roles="amq"/>
+ <!-- we need this otherwise ./artemis data imp wouldn't work -->
+ <permission type="manage" roles="amq"/>
+ </security-setting>
+ </security-settings>
+
+ <address-settings>
+ <!-- if you define auto-create on certain queues, management has to be auto-create -->
+ <address-setting match="activemq.management#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <!-- with -1 only the global-max-size is in use for limiting -->
+ <max-size-bytes>-1</max-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ <auto-create-jms-queues>true</auto-create-jms-queues>
+ <auto-create-jms-topics>true</auto-create-jms-topics>
+ </address-setting>
+ <!--default for catch all-->
+ <address-setting match="#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <!-- with -1 only the global-max-size is in use for limiting -->
+ <max-size-bytes>-1</max-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ <auto-create-jms-queues>true</auto-create-jms-queues>
+ <auto-create-jms-topics>true</auto-create-jms-topics>
+ </address-setting>
+ </address-settings>
+
+ <addresses>
+ <address name="DLQ">
+ <anycast>
+ <queue name="DLQ" />
+ </anycast>
+ </address>
+ <address name="ExpiryQueue">
+ <anycast>
+ <queue name="ExpiryQueue" />
+ </anycast>
+ </address>
+
+ <address name="bridgeQueue">
+ <anycast>
+ <queue name="bridgeQueue"/>
+ </anycast>
+ </address>
+
+ </addresses>
+
+
+ <!-- Uncomment the following if you want to use the Standard LoggingActiveMQServerPlugin pluging to log in events
+ <broker-plugins>
+ <broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
+ <property key="LOG_ALL_EVENTS" value="true"/>
+ <property key="LOG_CONNECTION_EVENTS" value="true"/>
+ <property key="LOG_SESSION_EVENTS" value="true"/>
+ <property key="LOG_CONSUMER_EVENTS" value="true"/>
+ <property key="LOG_DELIVERING_EVENTS" value="true"/>
+ <property key="LOG_SENDING_EVENTS" value="true"/>
+ <property key="LOG_INTERNAL_EVENTS" value="true"/>
+ </broker-plugin>
+ </broker-plugins>
+ -->
+
+ </core>
+</configuration>
diff --git a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/bridgeTransfer/BridgeTransferingTest.java b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/bridgeTransfer/BridgeTransferingTest.java
new file mode 100644
index 0000000..5b58857
--- /dev/null
+++ b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/bridgeTransfer/BridgeTransferingTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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 org.apache.activemq.artemis.tests.smoke.bridgeTransfer;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
+import org.apache.activemq.artemis.utils.Wait;
+import org.apache.qpid.jms.JmsConnectionFactory;
+import org.jboss.logging.Logger;
+import org.junit.After;
+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 BridgeTransferingTest extends SmokeTestBase {
+
+ public static final String SERVER_NAME_0 = "bridgeTransfer/serverA";
+ public static final String SERVER_NAME_1 = "bridgeTransfer/serverB";
+ private static final Logger logger = Logger.getLogger(BridgeTransferingTest.class);
+ private static final String JMX_SERVER_HOSTNAME = "localhost";
+ private static final int JMX_SERVER_PORT = 11099;
+
+ final String theprotocol;
+ // As the produces sends messages, a client will be killed every X messages. This is it!
+ final int killServerInterval;
+ final int numberOfMessages;
+ final int commitInterval;
+ final int messageSize;
+ final boolean killBothServers;
+ final int minlargeMessageSize;
+ Process serverProcess;
+ Process serverProcess2;
+
+ public BridgeTransferingTest(String protocol, int commitInterval, int killServerInterval, int numberOfMessages, int messageSize, int minlargeMessageSize, boolean killBothServers) {
+ this.theprotocol = protocol;
+ this.killServerInterval = killServerInterval;
+ this.messageSize = messageSize;
+ this.commitInterval = commitInterval;
+ this.numberOfMessages = numberOfMessages;
+ this.killBothServers = killBothServers;
+ this.minlargeMessageSize = minlargeMessageSize;
+ }
+
+ @Parameterized.Parameters(name = "protocol={0}, commitInterval={1}, killInterval={2}, numberOfMessages={3}, messageSize={4}, minLargeMessageSize={5}, KillBothServers={6}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][]{{"CORE", 200, 1000, 10000, 15_000, 5000, true}, {"CORE", 200, 1000, 10000, 15_000, 5000, false}});
+ }
+
+ public static ConnectionFactory createConnectionFactory(String protocol, String uri) {
+ if (protocol.toUpperCase().equals("OPENWIRE")) {
+ return new org.apache.activemq.ActiveMQConnectionFactory(uri);
+ } else 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);
+ }
+ }
+
+ @Before
+ public void before() throws Exception {
+ cleanupData(SERVER_NAME_0);
+ cleanupData(SERVER_NAME_1);
+ disableCheckThread();
+ serverProcess = startServer(SERVER_NAME_0, 0, 30000);
+ serverProcess2 = startServer(SERVER_NAME_1, 1, 30000);
+ }
+
+ @After
+ public void stopServers() throws Exception {
+ serverProcess2.destroyForcibly();
+ serverProcess.destroyForcibly();
+ }
+
+ @Test
+ public void testTransfer() throws Exception {
+ ConnectionFactory cf = createConnectionFactory(theprotocol, "tcp://localhost:61616");
+ ((ActiveMQConnectionFactory) cf).setMinLargeMessageSize(minlargeMessageSize);
+
+ String body;
+
+ {
+ StringBuffer buffer = new StringBuffer();
+ while (buffer.length() < messageSize) {
+ buffer.append(" ");
+ }
+ body = buffer.toString();
+ }
+
+ {
+ Connection connection = cf.createConnection();
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue("bridgeQueue");
+ MessageProducer producer = session.createProducer(queue);
+
+ int txElement = 0;
+ int killElement = 0;
+
+ for (int i = 0; i < numberOfMessages; i++) {
+ producer.send(session.createTextMessage(body + " " + i));
+
+ if (++txElement == commitInterval) {
+ logger.debug("Sent " + (i + 1) + " messages");
+ txElement = 0;
+ session.commit();
+ }
+
+ if (++killElement == killServerInterval) {
+ logger.debug("Killing server at " + (i + 1));
+ killElement = 0;
+ if (killBothServers) {
+ serverProcess.destroyForcibly();
+ Wait.assertFalse(serverProcess::isAlive);
+ }
+ serverProcess2.destroyForcibly();
+ Wait.assertFalse(serverProcess2::isAlive);
+ serverProcess2 = startServer(SERVER_NAME_1, 1, 30000);
+ if (killBothServers) {
+ serverProcess = startServer(SERVER_NAME_0, 0, 30000);
+ }
+ if (killBothServers) {
+ connection.close();
+ connection = cf.createConnection();
+ session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ queue = session.createQueue("bridgeQueue");
+ producer = session.createProducer(queue);
+ }
+ }
+ }
+
+ if (txElement > 0) {
+ session.commit();
+ }
+ }
+ ConnectionFactory cf2 = createConnectionFactory(theprotocol, "tcp://localhost:61617");
+ try (Connection connection = cf2.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("bridgeQueue");
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+
+ for (int i = 0; i < numberOfMessages; i++) {
+ if (i % 100 == 0) {
+ logger.debug("consuming " + i);
+ }
+ TextMessage message = (TextMessage) consumer.receive(5000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals(body + " " + i, message.getText());
+ }
+
+ Assert.assertNull(consumer.receiveNoWait());
+ }
+
+ }
+
+}