Skip to content

Commit b1bf961

Browse files
author
Tim Watson
committed
reduce duplication between per-queue and per-message ttl tests
1 parent 105b58e commit b1bf961

File tree

3 files changed

+186
-292
lines changed

3 files changed

+186
-292
lines changed

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

Lines changed: 28 additions & 153 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,20 @@
1+
// The contents of this file are subject to the Mozilla Public License
2+
// Version 1.1 (the "License"); you may not use this file except in
3+
// compliance with the License. You may obtain a copy of the License
4+
// at http://www.mozilla.org/MPL/
5+
//
6+
// Software distributed under the License is distributed on an "AS IS"
7+
// basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
8+
// the License for the specific language governing rights and
9+
// limitations under the License.
10+
//
11+
// The Original Code is RabbitMQ.
12+
//
13+
// The Initial Developer of the Original Code is VMware, Inc.
14+
// Copyright (c) 2007-2012 VMware, Inc. All rights reserved.
15+
//
16+
17+
118
package com.rabbitmq.client.test.functional;
219

320
import com.rabbitmq.client.*;
@@ -6,180 +23,38 @@
623

724
public class PerMessageTTL extends TTLHandling {
825

9-
@Override
10-
protected void createResources() throws IOException {
11-
super.createResources();
12-
declareAndBindQueue();
13-
this.channel.confirmSelect();
14-
}
26+
private Object sessionTTL;
1527

1628
@Override
17-
protected void releaseResources() throws IOException {
18-
super.releaseResources();
19-
}
20-
21-
public void testSupportedTTLTypes() throws IOException {
22-
Object[] args = { (((byte)200) & (0xff)), (short)200, 200, 200L };
23-
for (Object ttl : args) {
24-
try {
25-
publishAndSynchronise(MSG[0], ttl);
26-
} catch(Exception ex) {
27-
fail("Should be able to use " + ttl.getClass().getName() +
28-
" for basic.expiration: " + ex.getMessage());
29-
}
30-
}
31-
}
32-
33-
public void testTTLAllowZero() throws Exception {
34-
try {
35-
publish(MSG[0], (byte) 0);
36-
this.channel.waitForConfirmsOrDie();
37-
} catch (Exception e) {
38-
fail("Should be able to publish with basic.expiration set to zero: " +
39-
e.getMessage());
40-
}
41-
}
42-
43-
public void testPublishWithInvalidTTL() throws InterruptedException {
29+
public void testInvalidTypeUsedInTTL() throws Exception {
4430
try {
45-
publishAndSynchronise(MSG[0], "foobar");
46-
fail("Should not be able to set a non-long value for basic.expiration");
31+
super.testInvalidTypeUsedInTTL();
4732
} catch (IOException e) {
4833
checkShutdownSignal(AMQP.INTERNAL_ERROR, e);
4934
}
5035
}
5136

37+
@Override
5238
public void testTTLMustBePositive() throws Exception {
5339
try {
54-
publishAndSynchronise(MSG[0], -15);
55-
fail("Should not be able to set a negative value for basic.expiration");
40+
super.testTTLMustBePositive();
5641
} catch (IOException e) {
5742
checkShutdownSignal(AMQP.INTERNAL_ERROR, e);
5843
}
5944
}
6045

61-
public void testMessagesExpireWhenUsingBasicGet() throws Exception {
62-
publish(MSG[0], 200);
63-
Thread.sleep(1000);
64-
65-
String what = get();
66-
assertNull("expected message " + what + " to have been removed", what);
67-
}
68-
69-
public void testMultiplePublishAndGetWithExpiry() throws Exception {
70-
// this seems quite timing dependent - would it not be better
71-
// to test this by setting up a DLX and verifying that the
72-
// expired messages have been sent there instead?
73-
publish(MSG[0], 200);
74-
Thread.sleep(500);
75-
76-
publish(MSG[1], 200);
77-
Thread.sleep(100);
78-
79-
publish(MSG[2], 200);
80-
81-
assertEquals(MSG[1], get());
82-
assertEquals(MSG[2], get());
83-
assertNull(get());
84-
}
85-
86-
/*
87-
* Test get expiry for messages sent under a transaction
88-
*/
89-
public void testTransactionalPublishWithGet() throws Exception {
90-
closeChannel();
91-
openChannel();
92-
this.channel.txSelect();
93-
94-
publish(MSG[0], 100);
95-
Thread.sleep(150);
96-
97-
publish(MSG[1], 100);
98-
this.channel.txCommit();
99-
Thread.sleep(50);
100-
101-
assertEquals(MSG[0], get());
102-
Thread.sleep(80);
103-
104-
assertNull(get());
105-
}
106-
107-
/*
108-
* Test expiry of requeued messages
109-
*/
110-
public void testExpiryWithReQueue() throws Exception {
111-
publish(MSG[0], 100);
112-
Thread.sleep(50);
113-
114-
publish(MSG[1], 100);
115-
publish(MSG[2], 100);
116-
117-
expectBodyAndRemainingMessages(MSG[0], 2);
118-
expectBodyAndRemainingMessages(MSG[1], 1);
119-
120-
closeChannel();
121-
openChannel();
122-
123-
Thread.sleep(60);
124-
expectBodyAndRemainingMessages(MSG[1], 1);
125-
expectBodyAndRemainingMessages(MSG[2], 0);
126-
}
127-
128-
/*
129-
* Test expiry of requeued messages after being consumed instantly
130-
*/
131-
public void testExpiryWithReQueueAfterConsume() throws Exception {
132-
QueueingConsumer c = new QueueingConsumer(channel);
133-
channel.basicConsume(TTL_QUEUE_NAME, c);
134-
135-
publish(MSG[0], 100);
136-
assertNotNull(c.nextDelivery(100));
137-
138-
closeChannel();
139-
Thread.sleep(150);
140-
openChannel();
141-
142-
assertNull("Re-queued message not expired", get());
143-
}
144-
145-
public void testZeroTTLDelivery() throws Exception {
146-
// when there is no consumer, message should expire
147-
publish(MSG[0], 0);
148-
assertNull(get());
149-
150-
// when there is a consumer, message should be delivered
151-
QueueingConsumer c = new QueueingConsumer(channel);
152-
channel.basicConsume(TTL_QUEUE_NAME, c);
153-
publish(MSG[0], 0);
154-
QueueingConsumer.Delivery d = c.nextDelivery(100);
155-
assertNotNull(d);
156-
157-
// requeued messages should expire
158-
channel.basicReject(d.getEnvelope().getDeliveryTag(), true);
159-
assertNull(c.nextDelivery(100));
160-
}
161-
162-
private void publishAndSynchronise(String msg, Object expiration)
163-
throws IOException, InterruptedException {
164-
publish(msg, expiration);
165-
this.channel.basicQos(0);
166-
}
167-
168-
private void publish(String msg, Object exp) throws IOException {
46+
@Override
47+
protected void publish(String msg) throws IOException {
16948
basicPublishVolatile(msg.getBytes(), TTL_EXCHANGE, TTL_QUEUE_NAME,
17049
MessageProperties.TEXT_PLAIN
17150
.builder()
172-
.expiration(String.valueOf(exp))
51+
.expiration(String.valueOf(sessionTTL))
17352
.build());
17453
}
17554

176-
private void declareAndBindQueue() throws IOException {
177-
declareQueue(TTL_QUEUE_NAME);
178-
this.channel.queueBind(TTL_QUEUE_NAME, TTL_EXCHANGE, TTL_QUEUE_NAME);
179-
}
180-
181-
private AMQP.Queue.DeclareOk declareQueue(String name) throws IOException {
55+
@Override
56+
protected AMQP.Queue.DeclareOk declareQueue(String name, Object ttlValue) throws IOException {
57+
this.sessionTTL = ttlValue;
18258
return this.channel.queueDeclare(name, false, true, false, null);
18359
}
184-
18560
}

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

Lines changed: 13 additions & 130 deletions
Original file line numberDiff line numberDiff line change
@@ -18,59 +18,40 @@
1818
package com.rabbitmq.client.test.functional;
1919

2020
import com.rabbitmq.client.AMQP;
21-
import com.rabbitmq.client.GetResponse;
22-
import com.rabbitmq.client.QueueingConsumer;
23-
import com.rabbitmq.client.QueueingConsumer.Delivery;
24-
import com.rabbitmq.client.test.BrokerTestCase;
2521

2622
import java.io.IOException;
2723
import java.util.Collections;
2824
import java.util.Map;
2925

30-
/**
31-
*
32-
*/
3326
public class PerQueueTTL extends TTLHandling {
3427

35-
public void testCreateQueueTTLTypes() throws IOException {
36-
Object[] args = { (byte)200, (short)200, 200, 200L };
37-
for (Object ttl : args) {
38-
try {
39-
declareQueue(ttl);
40-
} catch(IOException ex) {
41-
fail("Should be able to use " + ttl.getClass().getName() +
42-
" for x-message-ttl");
43-
}
44-
}
45-
}
46-
47-
public void testTTLAllowZero() throws Exception {
48-
try {
49-
declareQueue(0);
50-
} catch (IOException e) {
51-
fail("Should be able to declare a queue with zero for x-message-ttl");
52-
}
53-
}
28+
private static final String TTL_ARG = "x-message-ttl";
5429

55-
public void testCreateQueueWithInvalidTTL() throws Exception {
30+
@Override
31+
public void testInvalidTypeUsedInTTL() throws Exception {
5632
try {
57-
declareQueue(TTL_INVALID_QUEUE_NAME, "foobar");
58-
fail("Should not be able to declare a queue with a non-long value for x-message-ttl");
33+
super.testInvalidTypeUsedInTTL();
5934
} catch (IOException e) {
6035
checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
6136
}
6237
}
6338

39+
@Override
6440
public void testTTLMustBePositive() throws Exception {
6541
try {
66-
declareQueue(TTL_INVALID_QUEUE_NAME, -10);
67-
fail("Should not be able to declare a queue with negative value for x-message-ttl");
42+
super.testTTLMustBePositive();
6843
} catch (IOException e) {
6944
checkShutdownSignal(AMQP.PRECONDITION_FAILED, e);
7045
}
7146
}
7247

73-
public void testQueueRedeclareEquivalence() throws Exception {
48+
@Override
49+
protected AMQP.Queue.DeclareOk declareQueue(String name, Object ttlValue) throws IOException {
50+
Map<String, Object> argMap = Collections.singletonMap(TTL_ARG, ttlValue);
51+
return this.channel.queueDeclare(name, false, true, false, argMap);
52+
}
53+
54+
public void testQueueReDeclareEquivalence() throws Exception {
7455
declareQueue(10);
7556
try {
7657
declareQueue(20);
@@ -97,102 +78,4 @@ public void testQueueReDeclareSemanticNonEquivalence() throws Exception {
9778
}
9879
}
9980

100-
/*
101-
* Test messages expire when using basic get.
102-
*/
103-
public void testPublishAndGetWithExpiry() throws Exception {
104-
declareAndBindQueue(200);
105-
106-
publish(MSG[0]);
107-
Thread.sleep(150);
108-
109-
publish(MSG[1]);
110-
Thread.sleep(100);
111-
112-
publish(MSG[2]);
113-
114-
assertEquals(MSG[1], get());
115-
assertEquals(MSG[2], get());
116-
}
117-
118-
/*
119-
* Test get expiry for messages sent under a transaction
120-
*/
121-
public void testTransactionalPublishWithGet() throws Exception {
122-
declareAndBindQueue(100);
123-
124-
this.channel.txSelect();
125-
126-
publish(MSG[0]);
127-
Thread.sleep(150);
128-
129-
publish(MSG[1]);
130-
this.channel.txCommit();
131-
Thread.sleep(50);
132-
133-
assertEquals(MSG[0], get());
134-
Thread.sleep(80);
135-
136-
assertNull(get());
137-
}
138-
139-
/*
140-
* Test expiry of requeued messages
141-
*/
142-
public void testExpiryWithRequeue() throws Exception {
143-
declareAndBindQueue(100);
144-
145-
publish(MSG[0]);
146-
Thread.sleep(50);
147-
publish(MSG[1]);
148-
publish(MSG[2]);
149-
150-
expectBodyAndRemainingMessages(MSG[0], 2);
151-
expectBodyAndRemainingMessages(MSG[1], 1);
152-
153-
closeChannel();
154-
openChannel();
155-
156-
Thread.sleep(60);
157-
expectBodyAndRemainingMessages(MSG[1], 1);
158-
expectBodyAndRemainingMessages(MSG[2], 0);
159-
}
160-
161-
/*
162-
* Test expiry of requeued messages after being consumed instantly
163-
*/
164-
public void testExpiryWithReQueueAfterConsume() throws Exception {
165-
declareAndBindQueue(100);
166-
QueueingConsumer c = new QueueingConsumer(channel);
167-
channel.basicConsume(TTL_QUEUE_NAME, c);
168-
169-
publish(MSG[0]);
170-
assertNotNull(c.nextDelivery(100));
171-
172-
closeChannel();
173-
Thread.sleep(150);
174-
openChannel();
175-
176-
assertNull("Requeued message not expired", get());
177-
}
178-
179-
public void testZeroTTLDelivery() throws Exception {
180-
declareAndBindQueue(0);
181-
182-
// when there is no consumer, message should expire
183-
publish(MSG[0]);
184-
assertNull(get());
185-
186-
// when there is a consumer, message should be delivered
187-
QueueingConsumer c = new QueueingConsumer(channel);
188-
channel.basicConsume(TTL_QUEUE_NAME, c);
189-
publish(MSG[0]);
190-
Delivery d = c.nextDelivery(100);
191-
assertNotNull(d);
192-
193-
// requeued messages should expire
194-
channel.basicReject(d.getEnvelope().getDeliveryTag(), true);
195-
assertNull(c.nextDelivery(100));
196-
}
197-
19881
}

0 commit comments

Comments
 (0)