Skip to content

Commit 37f02ec

Browse files
tonygtonyg
committed
merge bug20654 into default
2 parents 89041d5 + a93413e commit 37f02ec

File tree

3 files changed

+276
-0
lines changed

3 files changed

+276
-0
lines changed

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

Lines changed: 250 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,250 @@
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 at
4+
// 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 the
8+
// License for the specific language governing rights and limitations
9+
// under the License.
10+
//
11+
// The Original Code is RabbitMQ.
12+
//
13+
// The Initial Developers of the Original Code are LShift Ltd,
14+
// Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
15+
//
16+
// Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
17+
// Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
18+
// are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
19+
// Technologies LLC, and Rabbit Technologies Ltd.
20+
//
21+
// Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift
22+
// Ltd. Portions created by Cohesive Financial Technologies LLC are
23+
// Copyright (C) 2007-2009 Cohesive Financial Technologies
24+
// LLC. Portions created by Rabbit Technologies Ltd are Copyright
25+
// (C) 2007-2009 Rabbit Technologies Ltd.
26+
//
27+
// All Rights Reserved.
28+
//
29+
// Contributor(s): ______________________________________.
30+
//
31+
32+
package com.rabbitmq.client.test.functional;
33+
34+
import com.rabbitmq.client.AMQP;
35+
import com.rabbitmq.client.ReturnListener;
36+
import com.rabbitmq.client.GetResponse;
37+
38+
import java.io.IOException;
39+
import java.util.concurrent.atomic.AtomicBoolean;
40+
import java.util.Map;
41+
import java.util.HashMap;
42+
43+
public class AlternateExchange extends BrokerTestCase
44+
{
45+
46+
static private String[] resources = new String[]{"x","u","v"};
47+
static private String[] keys = new String[]{"x","u","v","z"};
48+
49+
static private boolean unrouted[] = new boolean[] {false, false, false};
50+
51+
private AtomicBoolean gotReturn = new AtomicBoolean();
52+
53+
/**
54+
* Determine which of the queues in our test configuration we
55+
* expect a message with routing key <code>key</code> to get
56+
* delivered to: the queue (if any) named <code>key</code>.
57+
*
58+
* @param key the routing key of the message
59+
* @return an array of booleans that when zipped with {@link
60+
* #resources} indicates whether the messages is expected to be
61+
* routed to the respective queue
62+
*/
63+
private static boolean[] expected(String key) {
64+
boolean[] expected = new boolean[resources.length];
65+
for (int i = 0; i < resources.length; i++) {
66+
expected[i] = resources[i].equals(key);
67+
}
68+
return expected;
69+
}
70+
71+
@Override protected void setUp() throws IOException {
72+
super.setUp();
73+
channel.setReturnListener(new ReturnListener() {
74+
public void handleBasicReturn(int replyCode,
75+
String replyText,
76+
String exchange,
77+
String routingKey,
78+
AMQP.BasicProperties properties,
79+
byte[] body)
80+
throws IOException {
81+
gotReturn.set(true);
82+
}
83+
});
84+
}
85+
86+
@Override protected void createResources() throws IOException {
87+
for (String q : resources) {
88+
channel.queueDeclare(q);
89+
}
90+
}
91+
92+
@Override protected void releaseResources() throws IOException {
93+
for (String q : resources) {
94+
channel.queueDelete(q);
95+
}
96+
}
97+
98+
/**
99+
* Declare an direct exchange <code>name</code> with an
100+
* alternate-exchange <code>ae</code> and bind the queue
101+
* <code>name</code> to it with a binding key of
102+
* <code>name</code>.
103+
*
104+
* @param name the name of the exchange to be created, and queue
105+
* to be bound
106+
* @param ae the name of the alternate-exchage
107+
*/
108+
protected void setupRouting(String name, String ae) throws IOException {
109+
Map<String, Object> args = new HashMap<String, Object>();
110+
if (ae != null) args.put("alternate-exchange", ae);
111+
channel.exchangeDeclare(name, "direct", false, false, false, args);
112+
channel.queueBind(name, name, name);
113+
}
114+
115+
protected void setupRouting() throws IOException {
116+
setupRouting("x", "u");
117+
setupRouting("u", "v");
118+
setupRouting("v", "x");
119+
}
120+
121+
protected void cleanup() throws IOException {
122+
for (String e : resources) {
123+
channel.exchangeDelete(e);
124+
}
125+
}
126+
127+
protected void publish(String key, boolean mandatory, boolean immediate)
128+
throws IOException {
129+
channel.basicPublish("x", key, mandatory, immediate, null,
130+
"ae-test".getBytes());
131+
}
132+
133+
protected void publish(String key) throws IOException {
134+
publish(key, false, false);
135+
}
136+
137+
/**
138+
* Perform an auto-acking 'basic.get' on each of the queues named
139+
* in {@link #resources} and check whether a message can be
140+
* retrieved when expected.
141+
*
142+
* @param expected an array of booleans that is zipped with {@link
143+
* #resources} and indicates whether a messages is expected
144+
* to be retrievable from the respective queue
145+
*/
146+
protected void checkGet(boolean[] expected) throws IOException {
147+
for (int i = 0; i < resources.length; i++) {
148+
String q = resources[i];
149+
GetResponse r = channel.basicGet(q, true);
150+
assertEquals("check " + q , expected[i], r != null);
151+
}
152+
}
153+
154+
/**
155+
* Test whether a message is routed as expected.
156+
*
157+
* We publish a message to exchange 'x' with a routing key of
158+
* <code>key</code>, check whether the message (actually, any
159+
* message) can be retrieved from the queues named in {@link
160+
* #resources} when expected, and whether a 'basic.return' is
161+
* received when expected.
162+
*
163+
* @param key the routing key of the message to be sent
164+
* @param mandatory whether the message should be marked as 'mandatory'
165+
* @param immediate whether the message should be marked as 'immediate'
166+
* @param expected indicates which queues we expect the message to
167+
* get routed to
168+
* @param ret whether a 'basic.return' is expected
169+
*
170+
* @see #checkGet(boolean[])
171+
*/
172+
protected void check(String key, boolean mandatory, boolean immediate,
173+
boolean[] expected, boolean ret)
174+
throws IOException {
175+
176+
gotReturn.set(false);
177+
publish(key, mandatory, immediate);
178+
checkGet(expected);
179+
assertEquals(ret, gotReturn.get());
180+
}
181+
182+
protected void check(String key, boolean[] expected, boolean ret)
183+
throws IOException {
184+
check(key, false, false, expected, ret);
185+
}
186+
187+
protected void check(String key, boolean mandatory, boolean immediate,
188+
boolean ret) throws IOException {
189+
check(key, mandatory, immediate, expected(key), ret);
190+
}
191+
192+
protected void check(String key, boolean ret) throws IOException {
193+
check(key, false, false, ret);
194+
}
195+
196+
/**
197+
* check various cases of missing AEs - we expect to see some
198+
* warnings in the server logs
199+
*/
200+
public void testMissing() throws IOException {
201+
setupRouting("x", "u");
202+
check("x", false); //no warning
203+
check("u", unrouted, false); //warning
204+
205+
setupRouting("u", "v");
206+
check("u", false); //no warning
207+
check("v", unrouted, false); //warning
208+
209+
setupRouting("v", null);
210+
check("v", false); //no warning
211+
check("z", unrouted, false); //no warning
212+
213+
cleanup();
214+
}
215+
216+
public void testAe() throws IOException {
217+
setupRouting();
218+
219+
for (String k : keys) {
220+
//ordinary
221+
check(k, false);
222+
//mandatory
223+
check(k, true, false, k.equals("z"));
224+
//immediate
225+
check(k, false, true, unrouted, true);
226+
}
227+
228+
//tx
229+
channel.txSelect();
230+
for (String k : keys) {
231+
publish(k);
232+
checkGet(unrouted);
233+
channel.txRollback();
234+
checkGet(unrouted);
235+
publish(k);
236+
checkGet(unrouted);
237+
channel.txCommit();
238+
checkGet(expected(k));
239+
}
240+
241+
cleanup();
242+
}
243+
244+
public void testCycleBreaking() throws IOException {
245+
setupRouting();
246+
check("z", false);
247+
cleanup();
248+
}
249+
250+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ public static TestSuite suite() {
5050
suite.addTestSuite(Bug20004Test.class);
5151
suite.addTestSuite(QosTests.class);
5252
suite.addTestSuite(Permissions.class);
53+
suite.addTestSuite(AlternateExchange.class);
5354
return suite;
5455
}
5556
}

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
package com.rabbitmq.client.test.functional;
3333

3434
import java.io.IOException;
35+
import java.util.Map;
36+
import java.util.HashMap;
3537

3638
import com.rabbitmq.client.AMQP;
3739
import com.rabbitmq.client.Channel;
@@ -223,6 +225,29 @@ public void with(String name) throws IOException {
223225
}});
224226
}
225227

228+
public void testAltExchConfiguration()
229+
throws IOException
230+
{
231+
runTest(false, false, false,
232+
createAltExchConfigTest("configure-me"));
233+
runTest(false, false, false,
234+
createAltExchConfigTest("configure-and-write-me"));
235+
runTest(false, true, false,
236+
createAltExchConfigTest("configure-and-read-me"));
237+
}
238+
239+
protected WithName createAltExchConfigTest(final String exchange)
240+
throws IOException
241+
{
242+
return new WithName() {
243+
public void with(String ae) throws IOException {
244+
Map<String, Object> args = new HashMap<String, Object>();
245+
args.put("alternate-exchange", ae);
246+
channel.exchangeDeclare(exchange, "direct", false, false, false, args);
247+
channel.exchangeDelete(exchange);
248+
}};
249+
}
250+
226251
protected void runConfigureTest(WithName test)
227252
throws IOException
228253
{

0 commit comments

Comments
 (0)