Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-225 Add Support for External Executor

Browse Source 
If no ExecutorService is provided when creating a new
connection, Rabbit uses a default fixed thread pool
executor with 5 threads.

You can now specify an executor on the
<rabbit:connection-factory/>.

Due to limitations in the Rabbit API,
this must either reference a ExecutorService (such as
one returned by Executors.new...Executor() methods, or
a Spring ThreadPoolTaskExecutor, such as one defined
using the <task:executor/> element.

The underlying API (newConnection()) with no arg
calls (newConnection(null)), so it is safe to
always use newConnection(exec), and not test for
null.
This commit is contained in:
Gary Russell
2012-04-20 14:58:33 -04:00
committed by Oleg Zhurakousky
parent 49cc759b5f
commit 5ea317798f
10 changed files with 206 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2011 the original author or authors.
* Copyright 2010-2012 the original author or authors.
*
* Licensed 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
@@ -21,6 +21,7 @@ import org.w3c.dom.Element;
/**
* @author Dave Syer
* @author Gary Russell
*/
class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
@@ -38,6 +39,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
private static final String PASSWORD_ATTRIBUTE = "password";
private static final String EXECUTOR_ATTRIBUTE = "executor";
@Override
protected Class<?> getBeanClass(Element element) {
return CachingConnectionFactory.class;
@@ -63,7 +66,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
NamespaceUtils.setValueIfAttributeDefined(builder, element, USER_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, PASSWORD_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, VIRTUAL_HOST_ATTRIBUTE);
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE);
}
}

29
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2012 the original author or authors.
*
* Licensed 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
@@ -16,14 +16,18 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
/**
* @author Dave Syer
* @author Gary Russell
*
*/
public abstract class AbstractConnectionFactory implements ConnectionFactory, DisposableBean {
@@ -36,6 +40,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
private final CompositeChannelListener channelListener = new CompositeChannelListener();
private volatile ExecutorService executorService;
/**
* Create a new SingleConnectionFactory for the given target ConnectionFactory.
* @param rabbitConnectionFactory the target ConnectionFactory
@@ -111,9 +117,28 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
this.channelListener.addDelegate(listener);
}
/**
* Provide an Executor for
* use by the Rabbit ConnectionFactory when creating connections.
* Can either be an ExecutorService or a Spring
* ThreadPoolTaskExecutor, as defined by a &lt;task:executor/&gt; element.
* @param executor The executor.
*/
public void setExecutor(Executor executor) {
boolean isExecutorService = executor instanceof ExecutorService;
boolean isThreadPoolTaskExecutor = executor instanceof ThreadPoolTaskExecutor;
Assert.isTrue(isExecutorService || isThreadPoolTaskExecutor);
if (isExecutorService) {
this.executorService = (ExecutorService) executor;
}
else {
this.executorService = ((ThreadPoolTaskExecutor) executor).getThreadPoolExecutor();
}
}
final protected Connection createBareConnection() {
try {
return new SimpleConnection(this.rabbitConnectionFactory.newConnection());
return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService));
} catch (IOException e) {
throw RabbitUtils.convertRabbitAccessException(e);
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -379,6 +379,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
MessageProperties messageProperties = messagePropertiesConverter.toMessageProperties(
properties, envelope, encoding);
Message reply = new Message(body, messageProperties);
if (logger.isTraceEnabled()) {
logger.trace("Message received " + reply);
}
try {
replyHandoff.put(reply);
} catch (InterruptedException e) {

14
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd
View File

@@ -774,6 +774,20 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="executor" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to an ExecutorService, or ThreadPoolTaskExecutor (as defined by a <task:executor/>
element). Passed to the Rabbit library when creating the connection. When not supplied, the
Rabbit library currently uses a fixed thread pool ExecutorService with 5 threads.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

34
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2011 the original author or authors.
* Copyright 2010-2012 the original author or authors.
*
* Licensed 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
@@ -15,16 +15,23 @@ package org.springframework.amqp.rabbit.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import java.util.concurrent.ExecutorService;
import org.junit.Before;
import org.junit.Test;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.xml.XmlBeanFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
*
* @author Dave Syer
* @author Gary Russell
*
*/
public final class ConnectionFactoryParserTests {
@@ -41,6 +48,7 @@ public final class ConnectionFactoryParserTests {
CachingConnectionFactory connectionFactory = beanFactory.getBean("kitchenSink", CachingConnectionFactory.class);
assertNotNull(connectionFactory);
assertEquals(10, connectionFactory.getChannelCacheSize());
assertNull(new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService"));
}
@Test
@@ -49,5 +57,27 @@ public final class ConnectionFactoryParserTests {
assertNotNull(connectionFactory);
assertEquals(10, connectionFactory.getChannelCacheSize());
}
@Test
public void testWithExecutor() throws Exception {
CachingConnectionFactory connectionFactory = beanFactory.getBean("withExecutor", CachingConnectionFactory.class);
assertNotNull(connectionFactory);
assertEquals(10, connectionFactory.getChannelCacheSize());
Object executor = new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService");
assertNotNull(executor);
ThreadPoolTaskExecutor exec = beanFactory.getBean("exec", ThreadPoolTaskExecutor.class);
assertSame(exec.getThreadPoolExecutor(), executor);
}
@Test
public void testWithExecutorService() throws Exception {
CachingConnectionFactory connectionFactory = beanFactory.getBean("withExecutorService", CachingConnectionFactory.class);
assertNotNull(connectionFactory);
assertEquals(10, connectionFactory.getChannelCacheSize());
Object executor = new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService");
assertNotNull(executor);
ExecutorService exec = beanFactory.getBean("execService", ExecutorService.class);
assertSame(exec, executor);
}
}

15
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java
View File

@@ -10,6 +10,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
@@ -28,7 +29,7 @@ public abstract class AbstractConnectionFactoryTests {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
final AtomicInteger called = new AtomicInteger(0);
AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
@@ -55,7 +56,7 @@ public abstract class AbstractConnectionFactoryTests {
assertEquals(0, called.get());
verify(mockConnection, atLeastOnce()).close();
verify(mockConnectionFactory, times(1)).newConnection();
verify(mockConnectionFactory, times(1)).newConnection((ExecutorService) null);
}
@@ -65,7 +66,7 @@ public abstract class AbstractConnectionFactoryTests {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
final AtomicInteger called = new AtomicInteger(0);
AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
@@ -93,7 +94,7 @@ public abstract class AbstractConnectionFactoryTests {
assertEquals(0, called.get());
verify(mockConnection, atLeastOnce()).close();
verify(mockConnectionFactory, times(1)).newConnection();
verify(mockConnectionFactory, times(1)).newConnection((ExecutorService) null);
}
@@ -104,7 +105,7 @@ public abstract class AbstractConnectionFactoryTests {
com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class);
com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection1).thenReturn(mockConnection2);
// simulate a dead connection
when(mockConnection1.isOpen()).thenReturn(false);
@@ -113,7 +114,7 @@ public abstract class AbstractConnectionFactoryTests {
Connection connection = connectionFactory.createConnection();
// the dead connection should be discarded
connection.createChannel(false);
verify(mockConnectionFactory, times(2)).newConnection();
verify(mockConnectionFactory, times(2)).newConnection((ExecutorService) null);
verify(mockConnection2, times(1)).createChannel();
connectionFactory.destroy();
@@ -129,7 +130,7 @@ public abstract class AbstractConnectionFactoryTests {
AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
connectionFactory.destroy();
verify(mockConnectionFactory, never()).newConnection();
verify(mockConnectionFactory, never()).newConnection((ExecutorService) null);
}
}

17
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -11,6 +11,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import junit.framework.Assert;
@@ -39,7 +40,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel);
when(mockChannel.isOpen()).thenReturn(true);
when(mockConnection.isOpen()).thenReturn(true);
@@ -72,7 +73,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
@@ -125,7 +126,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
Channel mockChannel2 = mock(Channel.class);
Channel mockChannel3 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2).thenReturn(mockChannel3);
when(mockConnection.isOpen()).thenReturn(true);
@@ -180,7 +181,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
when(mockConnection.isOpen()).thenReturn(true);
@@ -227,7 +228,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
when(mockConnection.isOpen()).thenReturn(true);
@@ -287,7 +288,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
Assert.assertNotSame(mockChannel1, mockChannel2);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
// You can't repeat 'when' statements for stubbing consecutive calls to
// the same method to returning different
// values.
@@ -354,7 +355,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockChannel.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(mockChannel);
@@ -383,7 +384,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
connectionFactory.destroy();
verify(mockConnection, atLeastOnce()).close();
verify(mockConnectionFactory).newConnection();
verify(mockConnectionFactory).newConnection((ExecutorService) null);
}
}

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java
View File

@@ -9,6 +9,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
@@ -33,7 +34,7 @@ public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(mockChannel);
@@ -60,7 +61,7 @@ public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests
connectionFactory.destroy();
verify(mockConnection, atLeastOnce()).close();
verify(mockConnectionFactory).newConnection();
verify(mockConnectionFactory).newConnection((ExecutorService) null);
}

85
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2011 the original author or authors.
* Copyright 2010-2012 the original author or authors.
*
* Licensed 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
@@ -17,17 +17,24 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.when;
import java.lang.reflect.Field;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
@@ -39,6 +46,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
@@ -46,6 +54,9 @@ import org.springframework.transaction.support.AbstractPlatformTransactionManage
import org.springframework.transaction.support.DefaultTransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.FieldCallback;
import org.springframework.util.ReflectionUtils.FieldFilter;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
@@ -319,6 +330,76 @@ public class RabbitTemplateIntegrationTests {
assertEquals(null, reply);
}
@Test
public void testAtomicSendAndReceiveExternalExecutor() throws Exception {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
final String execName = "make-sure-exec-passed-in";
exec.setBeanName(execName);
exec.afterPropertiesSet();
connectionFactory.setExecutor(exec);
final Field[] fields = new Field[1];
ReflectionUtils.doWithFields(RabbitTemplate.class, new FieldCallback() {
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
field.setAccessible(true);
fields[0] = field;
}
}, new FieldFilter() {
public boolean matches(Field field) {
return field.getName().equals("logger");
}
});
Log logger = Mockito.mock(Log.class);
when(logger.isTraceEnabled()).thenReturn(true);
final AtomicBoolean execConfiguredOk = new AtomicBoolean();
doAnswer(new Answer<Object>(){
public Object answer(InvocationOnMock invocation) throws Throwable {
String log = (String) invocation.getArguments()[0];
if (log.startsWith("Message received") &&
Thread.currentThread().getName().startsWith(execName)) {
execConfiguredOk.set(true);
}
return null;
}
}).when(logger).trace(Mockito.anyString());
final RabbitTemplate template = new RabbitTemplate(connectionFactory);
ReflectionUtils.setField(fields[0], template, logger);
template.setRoutingKey(ROUTE);
template.setQueue(ROUTE);
ExecutorService executor = Executors.newFixedThreadPool(1);
// Set up a consumer to respond to our producer
Future<Message> received = executor.submit(new Callable<Message>() {
public Message call() throws Exception {
Message message = null;
for (int i = 0; i < 10; i++) {
message = template.receive();
if (message != null) {
break;
}
Thread.sleep(100L);
}
assertNotNull("No message received", message);
template.send(message.getMessageProperties().getReplyTo(), message);
return message;
}
});
Message message = new Message("test-message".getBytes(), new MessageProperties());
Message reply = template.sendAndReceive(message);
assertEquals(new String(message.getBody()), new String(received.get(1000, TimeUnit.MILLISECONDS).getBody()));
assertNotNull("Reply is expected", reply);
assertEquals(new String(message.getBody()), new String(reply.getBody()));
// Message was consumed so nothing left on queue
reply = template.receive();
assertEquals(null, reply);
assertTrue(execConfiguredOk.get());
}
@Test
public void testAtomicSendAndReceiveWithRoutingKey() throws Exception {
final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());

26
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml
View File

@@ -1,8 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<rabbit:connection-factory id="kitchenSink" host="foo" virtual-host="/bar"
channel-cache-size="10" port="6888" username="user" password="password" />
@@ -11,4 +17,16 @@
<bean id="connectionFactory" class="com.rabbitmq.client.ConnectionFactory"/>
<rabbit:connection-factory id="withExecutor" host="foo" virtual-host="/bar"
channel-cache-size="10" port="6888" username="user" password="password"
executor="exec" />
<task:executor id="exec" />
<rabbit:connection-factory id="withExecutorService" host="foo" virtual-host="/bar"
channel-cache-size="10" port="6888" username="user" password="password"
executor="execService" />
<bean id="execService" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />
</beans>