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

AMQP-255 Fix Multi Confirm with > 1 Listener

Browse Source 
Rabbit can piggy-back confirms - for example, if seq 1, 2, 3 are
sent, it is possible to receive ack #3 with 'multiple' set. These
means 1, 2, and 3 are acked.

This worked fine with just one listener. However, if two or
more listeners (e.g. rabbit templates) are attached, only the
listener for ack #3 is notified (regardless of whether all
the acks belong to it).

The PublisherCallbackChannel maintains two maps: seq-to-listener
and listener-to-map(seq-to-correlation).

This fixes the problem by first finding all the listeners that
have pending confirms at or below the sequence number; and then
uses the second map to send the confirms to the appropriate
listener.

AMQP-255 Polishing

PR Comments
This commit is contained in:
Gary Russell
2012-07-18 15:11:22 -04:00
committed by Oleg Zhurakousky
parent 902b2abdde
commit 247bf0eb53
2 changed files with 109 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java
View File

@@ -17,12 +17,15 @@ package org.springframework.amqp.rabbit.support;
import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;
import org.apache.commons.logging.Log;
@@ -72,7 +75,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
private final Map<Listener, SortedMap<Long, PendingConfirm>> pendingConfirms
= new ConcurrentHashMap<PublisherCallbackChannel.Listener, SortedMap<Long,PendingConfirm>>();
private final Map<Long, Listener> listenerForSeq = new ConcurrentHashMap<Long, Listener>();
private final SortedMap<Long, Listener> listenerForSeq = new ConcurrentSkipListMap<Long, Listener>();
public PublisherCallbackChannelImpl(Channel delegate) {
this.delegate = delegate;
@@ -483,27 +486,55 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
}
private void processAck(long seq, boolean ack, boolean multiple) {
Listener listener = this.listenerForSeq.get(seq);
if (listener != null && listener.isConfirmListener()) {
if (multiple) {
Map<Long, PendingConfirm> headMap = this.pendingConfirms.get(listener).headMap(seq + 1);
synchronized(this.pendingConfirms) {
Iterator<Entry<Long, PendingConfirm>> iterator = headMap.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, PendingConfirm> entry = iterator.next();
iterator.remove();
listener.handleConfirm(entry.getValue(), ack);
if (multiple) {
/*
* Piggy-backed ack - extract all Listeners for this and earlier
* sequences. Then, for each Listener, handle each of it's acks.
*/
synchronized(this.pendingConfirms) {
Map<Long, Listener> involvedListeners = this.listenerForSeq.headMap(seq + 1);
// eliminate duplicates
Set<Listener> listeners = new HashSet<Listener>(involvedListeners.values());
for (Listener involvedListener : listeners) {
// find all unack'd confirms for this listener and handle them
SortedMap<Long, PendingConfirm> confirmsMap = this.pendingConfirms.get(involvedListener);
if (confirmsMap != null) {
Map<Long, PendingConfirm> confirms = confirmsMap.headMap(seq + 1);
Iterator<Entry<Long, PendingConfirm>> iterator = confirms.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, PendingConfirm> entry = iterator.next();
iterator.remove();
doHandleConfirm(ack, involvedListener, entry.getValue());
}
}
}
}
else {
}
else {
Listener listener = this.listenerForSeq.get(seq);
if (listener != null) {
PendingConfirm pendingConfirm = this.pendingConfirms.get(listener).remove(seq);
if (pendingConfirm != null) {
listener.handleConfirm(pendingConfirm, ack);
doHandleConfirm(ack, listener, pendingConfirm);
}
}
} else {
logger.error("No listener for seq:" + seq);
else {
logger.error("No listener for seq:" + seq);
}
}
}
private void doHandleConfirm(boolean ack, Listener listener, PendingConfirm pendingConfirm) {
try {
if (listener.isConfirmListener()) {
if (logger.isDebugEnabled()) {
logger.debug("Sending confirm " + pendingConfirm);
}
listener.handleConfirm(pendingConfirm, ack);
}
}
catch (Exception e) {
logger.error("Exception delivering confirm", e);
}
}

63
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -366,4 +366,67 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Collection<CorrelationData> unconfirmed = template.getUnconfirmed(0);
assertNull(unconfirmed);
}
/**
* Tests that piggy-backed confirms (multiple=true) are distributed to the proper
* template.
* @throws Exception
*/
@Test
public void testPublisherConfirmMultipleWithTwoListeners() throws Exception {
ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
PublisherCallbackChannelImpl callbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel()).thenReturn(callbackChannel);
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>(){
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
}}).when(mockChannel).getNextPublishSeqNo();
final RabbitTemplate template1 = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
final Set<String> confirms = new HashSet<String>();
final CountDownLatch latch1 = new CountDownLatch(1);
template1.setConfirmCallback(new ConfirmCallback() {
public void confirm(CorrelationData correlationData, boolean ack) {
if (ack) {
confirms.add(correlationData.getId() + "1");
latch1.countDown();
}
}
});
final RabbitTemplate template2 = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
final CountDownLatch latch2 = new CountDownLatch(1);
template2.setConfirmCallback(new ConfirmCallback() {
public void confirm(CorrelationData correlationData, boolean ack) {
if (ack) {
confirms.add(correlationData.getId() + "2");
latch2.countDown();
}
}
});
template1.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
template2.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def"));
template2.convertAndSend(ROUTE, (Object) "message", new CorrelationData("ghi"));
callbackChannel.handleAck(3, true);
assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS));
assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS));
Collection<CorrelationData> unconfirmed1 = template1.getUnconfirmed(0);
assertNull(unconfirmed1);
Collection<CorrelationData> unconfirmed2 = template2.getUnconfirmed(0);
assertNull(unconfirmed2);
assertTrue(confirms.contains("abc1"));
assertTrue(confirms.contains("def2"));
assertTrue(confirms.contains("ghi2"));
assertEquals(3, confirms.size());
}
}