Skip to content

Commit eb32431

Browse files
acogoluegnesacogoluegnes
committed
Use different queue and exchange names between tests
Avoid conflicts between tests in case the record is not completely gone.
1 parent ed73350 commit eb32431

File tree

3 files changed

+144
-142
lines changed

3 files changed

+144
-142
lines changed

‎pom.xml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -788,8 +788,10 @@
788788
<include>src/main/java/com/rabbitmq/client/impl/Environment.java</include>
789789
<include>src/main/java/com/rabbitmq/client/observation/**/*.java</include>
790790
<include>src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java</include>
791+
<include>src/test/java/com/rabbitmq/client/test/functional/DurableOnTransient.java</include>
791792
<include>src/test/java/com/rabbitmq/client/test/functional/QueueExclusivity.java</include>
792793
<include>src/test/java/com/rabbitmq/client/test/NettyTest.java</include>
794+
<include>src/test/java/com/rabbitmq/client/test/CloseInMainLoop.java</include>
793795
<include>src/test/java/com/rabbitmq/client/test/ProtocolVersionMismatch.java</include>
794796
<include>src/test/java/com/rabbitmq/client/test/TestUtils.java</include>
795797
</includes>

‎src/test/java/com/rabbitmq/client/test/CloseInMainLoop.java‎

Lines changed: 72 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
1-
// Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
1+
// Copyright (c) 2007-2025 Broadcom. All Rights Reserved.
2+
// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
23
//
34
// This software, the RabbitMQ Java client library, is triple-licensed under the
45
// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -12,21 +13,10 @@
1213
//
1314
// If you have any questions regarding licensing, please contact us at
1415
// [email protected].
15-
1616
package com.rabbitmq.client.test;
1717

1818
import static org.junit.jupiter.api.Assertions.assertTrue;
1919

20-
import java.io.IOException;
21-
import java.util.concurrent.CountDownLatch;
22-
import java.util.concurrent.Executors;
23-
import java.util.concurrent.TimeUnit;
24-
import java.util.concurrent.atomic.AtomicBoolean;
25-
26-
import javax.net.SocketFactory;
27-
28-
import org.junit.jupiter.api.Test;
29-
3020
import com.rabbitmq.client.AMQP;
3121
import com.rabbitmq.client.Channel;
3222
import com.rabbitmq.client.Command;
@@ -37,89 +27,106 @@
3727
import com.rabbitmq.client.impl.AMQConnection;
3828
import com.rabbitmq.client.impl.DefaultExceptionHandler;
3929
import com.rabbitmq.client.impl.SocketFrameHandler;
30+
import java.io.IOException;
31+
import java.util.concurrent.CountDownLatch;
32+
import java.util.concurrent.Executors;
33+
import java.util.concurrent.TimeUnit;
34+
import java.util.concurrent.atomic.AtomicBoolean;
35+
import javax.net.SocketFactory;
36+
import org.junit.jupiter.api.Test;
4037

41-
public class CloseInMainLoop extends BrokerTestCase{
42-
private final CountDownLatch closeLatch = new CountDownLatch(1);
43-
44-
private ConnectionFactory specialConnectionFactory() {
45-
ConnectionFactory f = TestUtils.connectionFactory();
46-
f.setExceptionHandler(new DefaultExceptionHandler(){
47-
@Override
48-
public void handleConsumerException(Channel channel,
49-
Throwable exception,
50-
Consumer consumer,
51-
String consumerTag,
52-
String methodName) {
53-
try {
54-
// TODO: change this to call 4-parameter close and make 6-parm one private
55-
((AMQConnection) channel.getConnection())
56-
.close(AMQP.INTERNAL_ERROR,
57-
"Internal error in Consumer " + consumerTag,
58-
false,
59-
exception,
60-
-1,
61-
false);
62-
} catch (Throwable e) {
63-
// Man, this clearly isn't our day.
64-
// TODO: Log the nested failure
65-
} finally {
66-
closeLatch.countDown();
67-
}
38+
public class CloseInMainLoop extends BrokerTestCase {
39+
private final CountDownLatch closeLatch = new CountDownLatch(1);
40+
41+
private ConnectionFactory specialConnectionFactory() {
42+
ConnectionFactory f = TestUtils.connectionFactory();
43+
f.setExceptionHandler(
44+
new DefaultExceptionHandler() {
45+
@Override
46+
public void handleConsumerException(
47+
Channel channel,
48+
Throwable exception,
49+
Consumer consumer,
50+
String consumerTag,
51+
String methodName) {
52+
try {
53+
// TODO: change this to call 4-parameter close and make 6-parm one private
54+
((AMQConnection) channel.getConnection())
55+
.close(
56+
AMQP.INTERNAL_ERROR,
57+
"Internal error in Consumer " + consumerTag,
58+
false,
59+
exception,
60+
-1,
61+
false);
62+
} catch (Throwable e) {
63+
// Man, this clearly isn't our day.
64+
// TODO: Log the nested failure
65+
} finally {
66+
closeLatch.countDown();
6867
}
68+
}
6969
});
70-
return f;
71-
}
70+
return f;
71+
}
7272

73-
class SpecialConnection extends AMQConnection{
73+
class SpecialConnection extends AMQConnection {
7474
private final AtomicBoolean validShutdown = new AtomicBoolean(false);
7575

76-
public boolean hadValidShutdown(){
77-
if(isOpen()) throw new IllegalStateException("hadValidShutdown called while connection is still open");
76+
public boolean hadValidShutdown() {
77+
if (isOpen())
78+
throw new IllegalStateException("hadValidShutdown called while connection is still open");
7879
return validShutdown.get();
7980
}
8081

8182
public SpecialConnection() throws Exception {
82-
super(specialConnectionFactory().params(Executors.newFixedThreadPool(1)),
83-
new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT)));
84-
this.start();
83+
super(
84+
specialConnectionFactory().params(Executors.newFixedThreadPool(1)),
85+
new SocketFrameHandler(
86+
SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT)));
87+
this.start();
8588
}
8689

8790
@Override
88-
public boolean processControlCommand(Command c) throws IOException{
89-
if(c.getMethod() instanceof AMQP.Connection.CloseOk) validShutdown.set(true);
91+
public boolean processControlCommand(Command c) throws IOException {
92+
if (c.getMethod() instanceof AMQP.Connection.CloseOk) validShutdown.set(true);
9093
return super.processControlCommand(c);
9194
}
9295
}
9396

94-
@Test public void closeOKNormallyReceived() throws Exception{
97+
@Test
98+
public void closeOKNormallyReceived() throws Exception {
9599
SpecialConnection connection = new SpecialConnection();
96100
connection.close(10_000);
97101
assertTrue(connection.hadValidShutdown());
98102
}
99103

100104
// The thrown runtime exception should get intercepted by the
101105
// consumer exception handler, and result in a clean shut down.
102-
@Test public void closeWithFaultyConsumer() throws Exception{
106+
@Test
107+
public void closeWithFaultyConsumer() throws Exception {
108+
String q = generateQueueName();
109+
String x = generateExchangeName();
103110
SpecialConnection connection = new SpecialConnection();
104111
Channel channel = connection.createChannel();
105-
channel.exchangeDeclare("x", "direct");
106-
channel.queueDeclare("q", false, false, false, null);
107-
channel.queueBind("q", "x", "k");
108-
109-
channel.basicConsume("q", true, new DefaultConsumer(channel){
110-
@Override
111-
public void handleDelivery(String consumerTag,
112-
Envelope envelope,
113-
AMQP.BasicProperties properties,
114-
byte[] body) {
112+
channel.exchangeDeclare(x, "direct");
113+
channel.queueDeclare(q, false, false, false, null);
114+
channel.queueBind(q, x, "k");
115+
116+
channel.basicConsume(
117+
q,
118+
true,
119+
new DefaultConsumer(channel) {
120+
@Override
121+
public void handleDelivery(
122+
String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
115123
throw new RuntimeException("I am a bad consumer");
116-
}
117-
});
124+
}
125+
});
118126

119-
channel.basicPublish("x", "k", null, new byte[10]);
127+
channel.basicPublish(x, "k", null, new byte[10]);
120128

121129
assertTrue(closeLatch.await(1000, TimeUnit.MILLISECONDS));
122130
assertTrue(connection.hadValidShutdown());
123131
}
124-
125132
}

‎src/test/java/com/rabbitmq/client/test/functional/DurableOnTransient.java‎

Lines changed: 70 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
1-
// Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
1+
// Copyright (c) 2007-2025 Broadcom. All Rights Reserved.
2+
// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
23
//
34
// This software, the RabbitMQ Java client library, is triple-licensed under the
45
// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -12,89 +13,81 @@
1213
//
1314
// If you have any questions regarding licensing, please contact us at
1415
// [email protected].
15-
16-
1716
package com.rabbitmq.client.test.functional;
1817

1918
import static org.junit.jupiter.api.Assertions.assertNotNull;
2019

21-
import java.io.IOException;
22-
23-
import org.junit.jupiter.api.Test;
24-
2520
import com.rabbitmq.client.GetResponse;
2621
import com.rabbitmq.client.MessageProperties;
22+
import java.io.IOException;
23+
import org.junit.jupiter.api.Test;
2724

28-
public class DurableOnTransient extends ClusteredTestBase
29-
{
30-
protected static final String Q = "SemiDurableBindings.DurableQueue";
31-
protected static final String X = "SemiDurableBindings.TransientExchange";
32-
33-
private GetResponse basicGet()
34-
throws IOException
35-
{
36-
return channel.basicGet(Q, true);
37-
}
38-
39-
private void basicPublish()
40-
throws IOException
41-
{
42-
channel.basicPublish(X, "",
43-
MessageProperties.PERSISTENT_TEXT_PLAIN,
44-
"persistent message".getBytes());
45-
}
46-
47-
protected void createResources() throws IOException {
48-
channel.exchangeDelete(X);
49-
// transient exchange
50-
channel.exchangeDeclare(X, "direct", false);
51-
52-
channel.queueDelete(Q);
53-
// durable queue
54-
channel.queueDeclare(Q, true, false, false, null);
55-
}
56-
57-
protected void releaseResources() throws IOException {
58-
channel.queueDelete(Q);
59-
channel.exchangeDelete(X);
60-
}
61-
62-
@Test public void bindDurableToTransient()
63-
throws IOException
64-
{
65-
channel.queueBind(Q, X, "");
66-
basicPublish();
67-
assertNotNull(basicGet());
68-
}
69-
70-
@Test public void semiDurableBindingRemoval() throws IOException {
71-
if (clusteredConnection != null) {
72-
deleteExchange("x");
73-
declareTransientTopicExchange("x");
74-
clusteredChannel.queueDelete("q");
75-
clusteredChannel.queueDeclare("q", true, false, false, null);
76-
channel.queueBind("q", "x", "k");
77-
78-
stopSecondary();
79-
boolean restarted = false;
80-
try {
81-
deleteExchange("x");
82-
83-
startSecondary();
84-
restarted = true;
85-
86-
declareTransientTopicExchange("x");
87-
88-
basicPublishVolatile("x", "k");
89-
assertDelivered("q", 0);
90-
91-
deleteQueue("q");
92-
deleteExchange("x");
93-
} finally {
94-
if (!restarted) {
95-
startSecondary();
96-
}
97-
}
25+
public class DurableOnTransient extends ClusteredTestBase {
26+
private String q;
27+
private String x;
28+
29+
private GetResponse basicGet() throws IOException {
30+
return channel.basicGet(q, true);
31+
}
32+
33+
private void basicPublish() throws IOException {
34+
channel.basicPublish(
35+
x, "", MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent message".getBytes());
36+
}
37+
38+
protected void createResources() throws IOException {
39+
x = generateExchangeName();
40+
q = generateQueueName();
41+
channel.exchangeDelete(x);
42+
// transient exchange
43+
channel.exchangeDeclare(x, "direct", false);
44+
45+
channel.queueDelete(q);
46+
// durable queue
47+
channel.queueDeclare(q, true, false, false, null);
48+
}
49+
50+
protected void releaseResources() throws IOException {
51+
channel.queueDelete(q);
52+
channel.exchangeDelete(x);
53+
}
54+
55+
@Test
56+
public void bindDurableToTransient() throws IOException {
57+
channel.queueBind(q, x, "");
58+
basicPublish();
59+
assertNotNull(basicGet());
60+
}
61+
62+
@Test
63+
public void semiDurableBindingRemoval() throws IOException {
64+
if (clusteredConnection != null) {
65+
String exchange = generateExchangeName();
66+
String queue = generateQueueName();
67+
declareTransientTopicExchange(exchange);
68+
clusteredChannel.queueDeclare(queue, true, false, false, null);
69+
channel.queueBind(queue, exchange, "k");
70+
71+
stopSecondary();
72+
boolean restarted = false;
73+
try {
74+
deleteExchange(exchange);
75+
76+
startSecondary();
77+
restarted = true;
78+
79+
declareTransientTopicExchange(exchange);
80+
81+
basicPublishVolatile(exchange, "k");
82+
assertDelivered(queue, 0);
83+
84+
deleteQueue(queue);
85+
deleteExchange(exchange);
86+
} finally {
87+
if (!restarted) {
88+
startSecondary();
9889
}
90+
}
9991
}
92+
}
10093
}

0 commit comments

Comments
 (0)