when(mockConnection.isOpen()).thenReturn(true);
PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel1);
PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel2);
when(mockConnection.createChannel()).thenReturn(channel1).thenReturn(channel2);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setPublisherConfirms(true);
ccf.setChannelCacheSize(3);
final RabbitTemplate template = new RabbitTemplate(ccf);
final AtomicBoolean confirmed = new AtomicBoolean();
template.setConfirmCallback(new ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
confirmed.set(true);
}
});
// Hold up the first thread so we get two channels
final CountDownLatch threadLatch = new CountDownLatch(1);
final CountDownLatch threadSentLatch = new CountDownLatch(1);
//Thread 1
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(new Runnable() {
@Override
public void run() {
template.execute(new ChannelCallback<Object>() {
@Override
public Object doInRabbit(Channel channel) throws Exception {
try {
threadLatch.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
template.doSend(channel, "", ROUTE,
new SimpleMessageConverter().toMessage("message", new MessageProperties()),
new CorrelationData("def"));
threadSentLatch.countDown();
return null;
}
});
}
});
// Thread 2
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); // channel y
threadLatch.countDown();
assertTrue(threadSentLatch.await(5, TimeUnit.SECONDS));
Collection<CorrelationData> unconfirmed = template.getUnconfirmed(-1);
assertEquals(2, unconfirmed.size());
Set<String> ids = new HashSet<String>();
Iterator<CorrelationData> iterator = unconfirmed.iterator();
ids.add(iterator.next().getId());
ids.add(iterator.next().getId());
assertTrue(ids.remove("abc"));
assertTrue(ids.remove("def"));
assertFalse(confirmed.get());
DirectFieldAccessor dfa = new DirectFieldAccessor(template);
Map<?, ?> pendingConfirms = (Map<?, ?>) dfa.getPropertyValue("pendingConfirms");
assertThat(pendingConfirms.size(), greaterThan(0)); // might use 2 or only 1 channel
exec.shutdown();
assertTrue(exec.awaitTermination(10, TimeUnit.SECONDS));
ccf.destroy();
assertEquals(0, pendingConfirms.size());
}