You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/06/06 07:02:57 UTC
[4/4] git commit: CAMEL-6869 merged the patch of Gerald
CAMEL-6869 merged the patch of Gerald
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/686d5fcc
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/686d5fcc
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/686d5fcc
Branch: refs/heads/master
Commit: 686d5fccac3e0308026548b58dfc90c31e41b5c9
Parents: 69b00a3 8029d0c
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Jun 6 10:58:16 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Jun 6 10:58:16 2014 +0800
----------------------------------------------------------------------
components/camel-rabbitmq/pom.xml | 17 ++-
.../component/rabbitmq/RabbitMQConsumer.java | 114 +++++++++++++----
.../component/rabbitmq/RabbitMQEndpoint.java | 36 ++++--
.../component/rabbitmq/RabbitMQProducer.java | 35 +++++-
.../rabbitmq/RabbitMQEndpointTest.java | 36 +++---
.../rabbitmq/RabbitMQReConnectionIntTest.java | 102 ++++++++++++++++
.../rabbitmq/RabbitMQSpringIntTest.java | 122 +++++++++++++++++++
.../rabbitmq/RabbitMQSpringIntTest-context.xml | 40 ++++++
8 files changed, 443 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
index 8a9b624,18ed62f..0f1d85f
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
@@@ -43,42 -48,56 +48,61 @@@ public class RabbitMQConsumer extends D
}
@Override
- protected void doStart() throws Exception {
- executor = endpoint.createExecutor();
- log.debug("Using executor {}", executor);
-
- conn = endpoint.connect(executor);
- log.debug("Using conn {}", conn);
+
- channel = conn.createChannel();
- log.debug("Using channel {}", channel);
+ public RabbitMQEndpoint getEndpoint() {
+ return (RabbitMQEndpoint) super.getEndpoint();
+ }
+ /**
+ * Open connection and channel
+ */
+ private void openConnectionAndChannel() throws IOException {
+ log.trace("Creating connection...");
+ this.conn = getEndpoint().connect(executor);
+ log.debug("Created connection: {}", conn);
+
+ log.trace("Creating channel...");
+ this.channel = conn.createChannel();
+ log.debug("Created channel: {}", channel);
-
- getEndpoint().declareExchangeAndQueue(channel);
- }
++ // setup the basicQos
+ if (endpoint.isPrefetchEnabled()) {
+ channel.basicQos(endpoint.getPrefetchSize(),
+ endpoint.getPrefetchCount(), endpoint.isPrefetchGlobal());
+ }
-
- channel.exchangeDeclare(endpoint.getExchangeName(),
- endpoint.getExchangeType(),
- endpoint.isDurable(),
- endpoint.isAutoDelete(),
- new HashMap<String, Object>());
-
- // need to make sure the queueDeclare is same with the exchange declare
- channel.queueDeclare(endpoint.getQueue(), endpoint.isDurable(), false,
- endpoint.isAutoDelete(), null);
- channel.queueBind(
- endpoint.getQueue(),
- endpoint.getExchangeName(),
- endpoint.getRoutingKey() == null ? "" : endpoint
- .getRoutingKey());
++ getEndpoint().declareExchangeAndQueue(channel);
++ }
+ /**
+ * If needed, create Exchange and Queue, then add message listener
+ */
+ private void addConsumer() throws IOException {
channel.basicConsume(endpoint.getQueue(), endpoint.isAutoAck(),
new RabbitConsumer(this, channel));
}
@Override
- protected void doStop() throws Exception {
+ protected void doStart() throws Exception {
+ executor = endpoint.createExecutor();
+ log.debug("Using executor {}", executor);
+ try {
+ openConnectionAndChannel();
+ addConsumer();
+ } catch (Exception e) {
+ // Open connection, and start message listener in background
+ Integer networkRecoveryInterval = getEndpoint().getNetworkRecoveryInterval();
- final long connectionRetryInterval= networkRecoveryInterval!=null && networkRecoveryInterval >0? networkRecoveryInterval :100L;
- startConsumerCallable=new StartConsumerCallable(connectionRetryInterval);
++ final long connectionRetryInterval = networkRecoveryInterval != null && networkRecoveryInterval > 0 ? networkRecoveryInterval : 100L;
++ startConsumerCallable = new StartConsumerCallable(connectionRetryInterval);
+ executor.submit(startConsumerCallable);
+ }
+ }
+
+ /**
+ * If needed, close Connection and Channel
+ */
+ private void closeConnectionAndChannel() throws IOException {
- if (startConsumerCallable!=null) {
++ if (startConsumerCallable != null) {
+ startConsumerCallable.stop();
+ }
if (channel != null) {
log.debug("Closing channel: {}", channel);
channel.close();
@@@ -183,4 -208,38 +213,38 @@@
}
+ /**
+ * Task in charge of opening connection and adding listener when consumer is started
- * and broker is not avaiblable.
++ * and broker is not available.
+ */
+ private class StartConsumerCallable implements Callable<Void> {
+ private final long connectionRetryInterval;
- private final AtomicBoolean running=new AtomicBoolean(true);
++ private final AtomicBoolean running = new AtomicBoolean(true);
+ public StartConsumerCallable(long connectionRetryInterval) {
+ this.connectionRetryInterval = connectionRetryInterval;
+ }
+ public void stop() {
+ running.set(false);
- RabbitMQConsumer.this.startConsumerCallable=null;
++ RabbitMQConsumer.this.startConsumerCallable = null;
+ }
+ @Override
+ public Void call() throws Exception {
- boolean connectionFailed=true;
++ boolean connectionFailed = true;
+ // Reconnection loop
+ while (running.get() && connectionFailed) {
+ try {
+ openConnectionAndChannel();
- connectionFailed=false;
++ connectionFailed = false;
+ } catch (Exception e) {
- log.debug("Connection failed, will retry in "+connectionRetryInterval+"ms", e);
++ log.debug("Connection failed, will retry in " + connectionRetryInterval + "ms", e);
+ Thread.sleep(connectionRetryInterval);
+ }
+ }
+ if (!connectionFailed) {
+ addConsumer();
+ }
+ stop();
+ return null;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index 9400935,7f87f53..e475819
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@@ -71,15 -73,7 +73,15 @@@ public class RabbitMQEndpoint extends D
private Boolean automaticRecoveryEnabled;
private Integer networkRecoveryInterval;
private Boolean topologyRecoveryEnabled;
-
+
+ //If it is true, prefetchSize, prefetchCount, prefetchGlobal will be used for basicOqs before starting RabbitMQConsumer
+ private boolean prefetchEnabled;
+ //Default in RabbitMq is 0.
+ private int prefetchSize;
+ private int prefetchCount;
+ //Default value in RabbitMQ is false.
+ private boolean prefetchGlobal;
-
++
public RabbitMQEndpoint() {
}
@@@ -134,7 -128,27 +136,27 @@@
}
}
- protected ConnectionFactory getOrCreateConnectionFactory() {
- /**
++ /**
+ * If needed, declare Exchange, declare Queue and bind them with Routing Key
+ */
+ public void declareExchangeAndQueue(Channel channel) throws IOException {
+ channel.exchangeDeclare(getExchangeName(),
+ getExchangeType(),
+ isDurable(),
+ isAutoDelete(),
+ new HashMap<String, Object>());
- if (getQueue()!=null) {
++ if (getQueue() != null) {
+ // need to make sure the queueDeclare is same with the exchange declare
+ channel.queueDeclare(getQueue(), isDurable(), false,
+ isAutoDelete(), null);
+ channel.queueBind(
+ getQueue(),
+ getExchangeName(),
+ getRoutingKey() == null ? "" : getRoutingKey());
+ }
+ }
+
+ private ConnectionFactory getOrCreateConnectionFactory() {
if (connectionFactory == null) {
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(getUsername());
@@@ -299,22 -313,22 +321,22 @@@
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
}
--
++
public void setBridgeEndpoint(boolean bridgeEndpoint) {
this.bridgeEndpoint = bridgeEndpoint;
}
--
++
public boolean isBridgeEndpoint() {
return bridgeEndpoint;
}
--
++
public void setAddresses(String addresses) {
Address[] addressArray = Address.parseAddresses(addresses);
if (addressArray.length > 0) {
this.addresses = addressArray;
}
}
--
++
public Address[] getAddresses() {
return addresses;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
index 7763423,f8596c9..2e22d3f
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQProducer.java
@@@ -57,6 -56,8 +56,8 @@@ public class RabbitMQProducer extends D
log.trace("Creating channel...");
this.channel = conn.createChannel();
log.debug("Created channel: {}", channel);
+
- getEndpoint().declareExchangeAndQueue(this.channel);
++ getEndpoint().declareExchangeAndQueue(this.channel);
}
@Override
@@@ -95,6 -114,10 +114,10 @@@
byte[] messageBodyBytes = exchange.getIn().getMandatoryBody(byte[].class);
AMQP.BasicProperties.Builder properties = buildProperties(exchange);
- if (channel==null) {
++ if (channel == null) {
+ // Open connection and channel lazily
+ openConnectionAndChannel();
+ }
channel.basicPublish(exchangeName, key, properties.build(), messageBodyBytes);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
index 86a7bcc,2db5005..afae40d
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
@@@ -109,7 -109,7 +109,7 @@@ public class RabbitMQEndpointTest exten
ThreadPoolExecutor executor = assertIsInstanceOf(ThreadPoolExecutor.class, endpoint.createExecutor());
assertEquals(20, executor.getCorePoolSize());
}
--
++
@Test
public void createEndpointWithAutoAckDisabled() throws Exception {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?autoAck=false", RabbitMQEndpoint.class);
@@@ -122,7 -122,7 +122,7 @@@
assertTrue(endpoint.isSingleton());
}
--
++
@Test
public void brokerEndpointAddressesSettings() throws Exception {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?addresses=server1:12345,server2:12345", RabbitMQEndpoint.class);
@@@ -133,7 -133,13 +133,13 @@@
private ConnectionFactory createConnectionFactory(String uri) {
RabbitMQEndpoint endpoint = context.getEndpoint(uri, RabbitMQEndpoint.class);
- return endpoint.getOrCreateConnectionFactory();
- try {
- endpoint.connect(Executors.newSingleThreadExecutor());
- } catch(IOException ioExc) {
- // Doesn't matter if RabbitMQ is not available
- log.debug("RabbitMQ not available", ioExc);
- }
++ try {
++ endpoint.connect(Executors.newSingleThreadExecutor());
++ } catch (IOException ioExc) {
++ // Doesn't matter if RabbitMQ is not available
++ log.debug("RabbitMQ not available", ioExc);
++ }
+ return endpoint.getConnectionFactory();
}
@Test
@@@ -157,17 -163,17 +163,17 @@@
@Test
public void testCreateConnectionFactoryCustom() throws Exception {
-- ConnectionFactory connectionFactory = createConnectionFactory("rabbitmq:localhost:1234/exchange"
- + "?username=userxxx"
- + "&password=passxxx"
- + "&connectionTimeout=123"
- + "&requestedChannelMax=456"
- + "&requestedFrameMax=789"
- + "&requestedHeartbeat=987"
- + "&sslProtocol=true"
- + "&automaticRecoveryEnabled=true"
- + "&networkRecoveryInterval=654"
- + "&topologyRecoveryEnabled=false");
- + "?username=userxxx"
- + "&password=passxxx"
- + "&connectionTimeout=123"
- + "&requestedChannelMax=456"
- + "&requestedFrameMax=789"
- + "&requestedHeartbeat=987"
- + "&sslProtocol=true"
- + "&automaticRecoveryEnabled=true"
- + "&networkRecoveryInterval=654"
- + "&topologyRecoveryEnabled=false");
++ ConnectionFactory connectionFactory = createConnectionFactory("rabbitmq:localhost:1234/exchange"
++ + "?username=userxxx"
++ + "&password=passxxx"
++ + "&connectionTimeout=123"
++ + "&requestedChannelMax=456"
++ + "&requestedFrameMax=789"
++ + "&requestedHeartbeat=987"
++ + "&sslProtocol=true"
++ + "&automaticRecoveryEnabled=true"
++ + "&networkRecoveryInterval=654"
++ + "&topologyRecoveryEnabled=false");
assertEquals("localhost", connectionFactory.getHost());
assertEquals(1234, connectionFactory.getPort());
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
index 0000000,512aed4..302440c
mode 000000,100644..100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQReConnectionIntTest.java
@@@ -1,0 -1,83 +1,102 @@@
++/**
++ * 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.camel.component.rabbitmq;
+
++import java.net.ConnectException;
++import java.util.concurrent.TimeUnit;
++
+ import com.rabbitmq.client.AlreadyClosedException;
-import org.apache.camel.*;
++
++import org.apache.camel.CamelExecutionException;
++import org.apache.camel.Endpoint;
++import org.apache.camel.EndpointInject;
++import org.apache.camel.Produce;
++import org.apache.camel.ProducerTemplate;
+ import org.apache.camel.builder.RouteBuilder;
+ import org.apache.camel.component.mock.MockEndpoint;
+ import org.apache.camel.test.junit4.CamelTestSupport;
+ import org.junit.Test;
+
-import java.net.ConnectException;
-import java.util.concurrent.TimeUnit;
-
+ /**
+ * Integration test to check that RabbitMQ Endpoint is able to reconnect to broker when broker
+ * is not avaibable.
+ * <ul>
+ * <li>Stop the broker</li>
+ * <li>Run the test: the producer complains it can not send messages, the consumer is silent</li>
+ * <li>Start the broker: the producer sends messages, and the consumer receives messages</li>
+ * <li>Stop the broker: the producer complains it can not send messages, the consumer is silent</li>
+ * <li>Start the broker: the producer sends messages, and the consumer receives messages</li>
+ * </ul>
+ */
+ public class RabbitMQReConnectionIntTest extends CamelTestSupport {
+ private static final String EXCHANGE = "ex3";
+
+ @Produce(uri = "direct:rabbitMQ")
+ protected ProducerTemplate directProducer;
+
- @EndpointInject(uri = "rabbitmq:localhost:5672/" + EXCHANGE + "?username=cameltest&password=cameltest" +
- "&queue=q3&routingKey=rk3" +
- "&automaticRecoveryEnabled=true" +
- "&requestedHeartbeat=1000" +
- "&connectionTimeout=5000")
++ @EndpointInject(uri = "rabbitmq:localhost:5672/" + EXCHANGE + "?username=cameltest&password=cameltest"
++ + "&queue=q3&routingKey=rk3" + "&automaticRecoveryEnabled=true"
++ + "&requestedHeartbeat=1000" + "&connectionTimeout=5000")
+ private Endpoint rabbitMQEndpoint;
+
+ @EndpointInject(uri = "mock:producing")
+ private MockEndpoint producingMockEndpoint;
+
+ @EndpointInject(uri = "mock:consuming")
+ private MockEndpoint consumingMockEndpoint;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ from("direct:rabbitMQ")
+ .id("producingRoute")
+ .onException(AlreadyClosedException.class, ConnectException.class)
+ .maximumRedeliveries(10)
+ .redeliveryDelay(500L)
+ .end()
+ .log("Sending message")
+ .inOnly(rabbitMQEndpoint)
+ .to(producingMockEndpoint);
+ from(rabbitMQEndpoint)
+ .id("consumingRoute")
+ .log("Receiving message")
+ .to(consumingMockEndpoint);
+ }
+ };
+ }
+
+ @Test
+ public void testSendEndReceive() throws Exception {
+ int nbMessages = 50;
+ int failedMessages = 0;
+ for (int i = 0; i < nbMessages; i++) {
+ try {
+ directProducer.sendBodyAndHeader("Message #" + i, RabbitMQConstants.ROUTING_KEY, "rk3");
+ } catch (CamelExecutionException e) {
+ log.debug("Can not send message", e);
+ failedMessages++;
+ }
+ Thread.sleep(500L);
+ }
+ producingMockEndpoint.expectedMessageCount(nbMessages - failedMessages);
+ consumingMockEndpoint.expectedMessageCount(nbMessages - failedMessages);
+ assertMockEndpointsSatisfied(5, TimeUnit.SECONDS);
+ }
+ }
http://git-wip-us.apache.org/repos/asf/camel/blob/686d5fcc/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQSpringIntTest.java
----------------------------------------------------------------------
diff --cc components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQSpringIntTest.java
index 0000000,baa4bde..6119082
mode 000000,100644..100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQSpringIntTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQSpringIntTest.java
@@@ -1,0 -1,102 +1,122 @@@
++/**
++ * 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.camel.component.rabbitmq;
+
-import com.rabbitmq.client.*;
++import java.io.IOException;
++
++import com.rabbitmq.client.AMQP;
++import com.rabbitmq.client.Channel;
++import com.rabbitmq.client.Connection;
++import com.rabbitmq.client.ConnectionFactory;
++import com.rabbitmq.client.DefaultConsumer;
++import com.rabbitmq.client.Envelope;
++
+ import org.apache.camel.Produce;
+ import org.apache.camel.ProducerTemplate;
+ import org.apache.camel.test.spring.CamelSpringJUnit4ClassRunner;
+ import org.junit.After;
+ import org.junit.Before;
+ import org.junit.Test;
+ import org.junit.runner.RunWith;
+ import org.springframework.beans.factory.annotation.Autowired;
+ import org.springframework.test.context.ContextConfiguration;
-
-import java.io.IOException;
-
+ import static org.junit.Assert.assertEquals;
-
+ /**
+ * Test RabbitMQ component with Spring DSL
+ */
+ @RunWith(CamelSpringJUnit4ClassRunner.class)
+ @ContextConfiguration("RabbitMQSpringIntTest-context.xml")
+ public class RabbitMQSpringIntTest {
+ @Produce(uri = "direct:rabbitMQ")
+ protected ProducerTemplate template;
+ @Autowired
+ private ConnectionFactory connectionFactory;
+ private Connection connection;
+ private Channel channel;
+
+ private Connection openConnection() throws IOException {
+ if (connection == null) {
+ connection = connectionFactory.newConnection();
+ }
+ return connection;
+ }
+
+ private Channel openChannel() throws IOException {
+ if (channel == null) {
+ channel = openConnection().createChannel();
+ }
+ return channel;
+ }
+
+ @Before
+ public void bindQueueExchange() throws IOException {
+ openChannel();
+ channel.exchangeDeclare("ex2", "direct", true, false, null);
+ channel.queueDeclare("q2", true, false, false, null);
+ channel.queueBind("q2", "ex2", "rk2");
+ }
+
+ @After
+ public void closeConnection() {
+ if (channel != null) {
+ try {
+ channel.close();
+ } catch (IOException e) {
+ }
+ }
+ if (connection != null) {
+ try {
+ connection.close();
+ } catch (IOException e) {
+ }
+ }
+ }
+
- private static class LastDeliveryConsumer extends DefaultConsumer {
++ private static final class LastDeliveryConsumer extends DefaultConsumer {
+ private byte[] lastBody;
+
+ private LastDeliveryConsumer(Channel channel) {
+ super(channel);
+ }
+
+ @Override
+ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+ lastBody = body;
+ super.handleDelivery(consumerTag, envelope, properties, body);
+ }
+
+ public byte[] getLastBody() {
+ return lastBody;
+ }
+ }
+
+ @Test
+ public void testSendCsutomConnectionFactory() throws Exception {
+ String body = "Hello Rabbit";
+ template.sendBodyAndHeader(body, RabbitMQConstants.ROUTING_KEY, "rk2");
+
+ openChannel();
+ LastDeliveryConsumer consumer = new LastDeliveryConsumer(channel);
+ channel.basicConsume("q2", true, consumer);
+ int i = 10;
+ while (consumer.getLastBody() == null && i > 0) {
+ Thread.sleep(1000L);
+ i--;
+ }
+ assertEquals(body, new String(consumer.getLastBody()));
+ }
+ }