1+ /*
2+ * Licensed to the Apache Software Foundation (ASF) under one or more
3+ * contributor license agreements. See the NOTICE file distributed with
4+ * this work for additional information regarding copyright ownership.
5+ * The ASF licenses this file to You under the Apache License, Version 2.0
6+ * (the "License"); you may not use this file except in compliance with
7+ * the License. You may obtain a copy of the License at
8+ *
9+ * http://www.apache.org/licenses/LICENSE-2.0
10+ *
11+ * Unless required by applicable law or agreed to in writing, software
12+ * distributed under the License is distributed on an "AS IS" BASIS,
13+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+ * See the License for the specific language governing permissions and
15+ * limitations under the License.
16+ */
17+ package org .apache .activemq .artemis .tests .integration .amqp .connect ;
18+
19+ import org .apache .activemq .artemis .core .config .amqpBrokerConnectivity .AMQPBrokerConnectConfiguration ;
20+ import org .apache .activemq .artemis .core .config .amqpBrokerConnectivity .AMQPMirrorBrokerConnectionElement ;
21+ import org .apache .activemq .artemis .core .server .ActiveMQServer ;
22+ import org .apache .activemq .artemis .core .server .Queue ;
23+ import org .apache .activemq .artemis .tests .integration .amqp .AmqpClientTestSupport ;
24+ import org .apache .activemq .artemis .tests .util .CFUtil ;
25+ import org .apache .activemq .artemis .utils .Wait ;
26+ import org .junit .jupiter .api .Test ;
27+
28+ import javax .jms .*;
29+
30+ import static org .junit .jupiter .api .Assertions .*;
31+
32+ public class AMQSquareMirroringTest extends AmqpClientTestSupport {
33+
34+ protected static final int AMQP_PORT_2 = 5673 ;
35+ protected static final int AMQP_PORT_3 = 5674 ;
36+ protected static final int AMQP_PORT_4 = 5675 ;
37+
38+ ActiveMQServer server_2 ;
39+ ActiveMQServer server_3 ;
40+ ActiveMQServer server_4 ;
41+
42+ @ Override
43+ protected ActiveMQServer createServer () throws Exception {
44+ return createServer (AMQP_PORT , false );
45+ }
46+
47+ @ Test
48+ public void testSquare () throws Exception {
49+ server_2 = createServer (AMQP_PORT_2 , false );
50+ server_3 = createServer (AMQP_PORT_3 , false );
51+ server_4 = createServer (AMQP_PORT_4 , false );
52+
53+ // name the servers, for convenience during debugging
54+ server .getConfiguration ().setName ("1" );
55+ server_2 .getConfiguration ().setName ("2" );
56+ server_3 .getConfiguration ().setName ("3" );
57+ server_4 .getConfiguration ().setName ("4" );
58+
59+ /**
60+ * 1 <----> 2
61+ * ^ ^
62+ * | |
63+ * v v
64+ * 4 <----> 3
65+ */
66+
67+ {
68+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration ("to2" , "tcp://localhost:" + AMQP_PORT_2 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
69+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
70+ server .getConfiguration ().addAMQPConnection (amqpConnection );
71+ amqpConnection = new AMQPBrokerConnectConfiguration ("to4" , "tcp://localhost:" + AMQP_PORT_4 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
72+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ().setForward (false ));
73+ server .getConfiguration ().addAMQPConnection (amqpConnection );
74+ }
75+
76+ {
77+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration ("to1" , "tcp://localhost:" + AMQP_PORT ).setRetryInterval (100 ).setReconnectAttempts (-1 );
78+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
79+ server_2 .getConfiguration ().addAMQPConnection (amqpConnection );
80+ amqpConnection = new AMQPBrokerConnectConfiguration ("to3" , "tcp://localhost:" + AMQP_PORT_3 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
81+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
82+ server_2 .getConfiguration ().addAMQPConnection (amqpConnection );
83+ }
84+
85+ {
86+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration ("to2" , "tcp://localhost:" + AMQP_PORT_2 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
87+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
88+ server_3 .getConfiguration ().addAMQPConnection (amqpConnection );
89+ amqpConnection = new AMQPBrokerConnectConfiguration ("to4" , "tcp://localhost:" + AMQP_PORT_4 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
90+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
91+ server_3 .getConfiguration ().addAMQPConnection (amqpConnection );
92+ }
93+
94+ {
95+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration ("to1" , "tcp://localhost:" + AMQP_PORT ).setRetryInterval (100 ).setReconnectAttempts (-1 );
96+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
97+ server_4 .getConfiguration ().addAMQPConnection (amqpConnection );
98+ amqpConnection = new AMQPBrokerConnectConfiguration ("to3" , "tcp://localhost:" + AMQP_PORT_3 ).setRetryInterval (100 ).setReconnectAttempts (-1 );
99+ amqpConnection .addElement (new AMQPMirrorBrokerConnectionElement ());
100+ server_4 .getConfiguration ().addAMQPConnection (amqpConnection );
101+ }
102+
103+ server .start ();
104+ server_2 .start ();
105+ server_3 .start ();
106+ server_4 .start ();
107+
108+ createAddressAndQueues (server );
109+ Wait .assertTrue (() -> server .locateQueue (getQueueName ()) != null );
110+ Wait .assertTrue (() -> server_2 .locateQueue (getQueueName ()) != null );
111+ Wait .assertTrue (() -> server_3 .locateQueue (getQueueName ()) != null );
112+ Wait .assertTrue (() -> server_4 .locateQueue (getQueueName ()) != null );
113+
114+ Queue q1 = server .locateQueue (getQueueName ());
115+ assertNotNull (q1 );
116+
117+ Queue q2 = server .locateQueue (getQueueName ());
118+ assertNotNull (q2 );
119+
120+ Queue q3 = server .locateQueue (getQueueName ());
121+ assertNotNull (q3 );
122+
123+ Queue q4 = server .locateQueue (getQueueName ());
124+ assertNotNull (q4 );
125+
126+ ConnectionFactory factory = CFUtil .createConnectionFactory ("AMQP" , "tcp://localhost:" + AMQP_PORT +"?amqp.idleTimeout=-1" );
127+ ConnectionFactory factory2 = CFUtil .createConnectionFactory ("AMQP" , "tcp://localhost:" + AMQP_PORT_2 +"?amqp.idleTimeout=-1" );
128+ ConnectionFactory factory3 = CFUtil .createConnectionFactory ("AMQP" , "tcp://localhost:" + AMQP_PORT_3 +"?amqp.idleTimeout=-1" );
129+ ConnectionFactory factory4 = CFUtil .createConnectionFactory ("AMQP" , "tcp://localhost:" + AMQP_PORT_4 +"?amqp.idleTimeout=-1" );
130+
131+ try (Connection conn = factory .createConnection ()) {
132+ Session session = conn .createSession ();
133+ MessageProducer producer = session .createProducer (session .createQueue (getQueueName ()));
134+ for (int i = 0 ; i < 40 ; i ++) {
135+ producer .send (session .createTextMessage ("message " + i ));
136+ }
137+ }
138+
139+ Thread .sleep (1_000 ); // some time to allow eventual loops
140+
141+ Wait .assertEquals (40L , q1 ::getMessageCount , 1000 , 100 );
142+ Wait .assertEquals (40L , q2 ::getMessageCount , 1000 , 100 );
143+ Wait .assertEquals (40L , q3 ::getMessageCount , 1000 , 100 );
144+ Wait .assertEquals (40L , q4 ::getMessageCount , 1000 , 100 );
145+
146+ try (Connection conn = factory .createConnection ()) {
147+ Session session = conn .createSession ();
148+ conn .start ();
149+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
150+ for (int i = 0 ; i < 10 ; i ++) {
151+ TextMessage message = (TextMessage ) consumer .receive (1000 );
152+ assertNotNull (message );
153+ assertEquals ("message " + i , message .getText ());
154+ }
155+ consumer .close ();
156+ }
157+
158+ Wait .assertEquals (30L , q1 ::getMessageCount , 1000 , 100 );
159+ Wait .assertEquals (30L , q2 ::getMessageCount , 1000 , 100 );
160+ Wait .assertEquals (30L , q3 ::getMessageCount , 1000 , 100 );
161+ Wait .assertEquals (30L , q4 ::getMessageCount , 1000 , 100 );
162+
163+ try (Connection conn = factory2 .createConnection ()) {
164+ Session session = conn .createSession ();
165+ conn .start ();
166+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
167+ for (int i = 10 ; i < 20 ; i ++) {
168+ TextMessage message = (TextMessage ) consumer .receive (1000 );
169+ assertNotNull (message );
170+ assertEquals ("message " + i , message .getText ());
171+ }
172+ consumer .close ();
173+ }
174+
175+ Wait .assertEquals (20L , q1 ::getMessageCount , 1000 , 100 );
176+ Wait .assertEquals (20L , q2 ::getMessageCount , 1000 , 100 );
177+ Wait .assertEquals (20L , q3 ::getMessageCount , 1000 , 100 );
178+ Wait .assertEquals (20L , q4 ::getMessageCount , 1000 , 100 );
179+
180+ try (Connection conn = factory3 .createConnection ()) {
181+ Session session = conn .createSession ();
182+ conn .start ();
183+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
184+ for (int i = 20 ; i < 30 ; i ++) {
185+ TextMessage message = (TextMessage ) consumer .receive (1000 );
186+ assertNotNull (message );
187+ assertEquals ("message " + i , message .getText ());
188+ }
189+ consumer .close ();
190+ }
191+
192+ Wait .assertEquals (10L , q1 ::getMessageCount , 1000 , 100 );
193+ Wait .assertEquals (10L , q2 ::getMessageCount , 1000 , 100 );
194+ Wait .assertEquals (10L , q3 ::getMessageCount , 1000 , 100 );
195+ Wait .assertEquals (10L , q4 ::getMessageCount , 1000 , 100 );
196+
197+ try (Connection conn = factory4 .createConnection ()) {
198+ Session session = conn .createSession ();
199+ conn .start ();
200+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
201+ for (int i = 30 ; i < 40 ; i ++) {
202+ TextMessage message = (TextMessage ) consumer .receive (1000 );
203+ assertNotNull (message );
204+ assertEquals ("message " + i , message .getText ());
205+ }
206+ consumer .close ();
207+ }
208+
209+ Wait .assertEquals (0L , q1 ::getMessageCount , 1000 , 100 );
210+ Wait .assertEquals (0L , q2 ::getMessageCount , 1000 , 100 );
211+ Wait .assertEquals (0L , q3 ::getMessageCount , 1000 , 100 );
212+ Wait .assertEquals (0L , q4 ::getMessageCount , 1000 , 100 );
213+
214+ try (Connection conn = factory .createConnection ()) {
215+ Session session = conn .createSession ();
216+ conn .start ();
217+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
218+ assertNull (consumer .receiveNoWait ());
219+ consumer .close ();
220+ }
221+
222+ try (Connection conn = factory2 .createConnection ()) {
223+ Session session = conn .createSession ();
224+ conn .start ();
225+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
226+ assertNull (consumer .receiveNoWait ());
227+ consumer .close ();
228+ }
229+
230+ try (Connection conn = factory3 .createConnection ()) {
231+ Session session = conn .createSession ();
232+ conn .start ();
233+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
234+ assertNull (consumer .receiveNoWait ());
235+ consumer .close ();
236+ }
237+
238+ try (Connection conn = factory4 .createConnection ()) {
239+ Session session = conn .createSession ();
240+ conn .start ();
241+ MessageConsumer consumer = session .createConsumer (session .createQueue (getQueueName ()));
242+ assertNull (consumer .receiveNoWait ());
243+ consumer .close ();
244+ }
245+
246+ }
247+ }
0 commit comments