cachingConnectionFactory.destroy();
}
@Test
public void testAtomicSendAndReceiveExternalExecutor() throws Exception {
final CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
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() {
@Override
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
field.setAccessible(true);
fields[0] = field;
}
}, new FieldFilter() {
@Override
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>(){
@Override
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>() {
@Override
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());
connectionFactory.destroy();
}