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

AMQP-532: Fix Pub.Conf. Nacks on close()

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-532

Properly use the iterator to remove the entries.
This commit is contained in:
Gary Russell
2015-09-11 16:03:28 -04:00
committed by Artem Bilan
parent 5eac90369d
commit 4e129e06a7
2 changed files with 62 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -633,14 +633,15 @@ public class PublisherCallbackChannelImpl
private synchronized void generateNacksForPendingAcks(String cause) {
for (Entry<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
Listener listener = entry.getKey();
for (Entry<Long, PendingConfirm> confirmEntry : entry.getValue().entrySet()) {
try {
confirmEntry.getValue().setCause(cause);
handleNack(confirmEntry.getKey(), false);
}
catch (IOException e) {
logger.error("Error delivering Nack afterShutdown", e);
Iterator<Entry<Long, PendingConfirm>> iterator = entry.getValue().entrySet().iterator();
while(iterator.hasNext()) {
Entry<Long, PendingConfirm> confirmEntry = iterator.next();
confirmEntry.getValue().setCause(cause);
if (logger.isDebugEnabled()) {
logger.debug(this.toString() + " PC:Nack:(close):" + confirmEntry.getKey());
}
processAck(confirmEntry.getKey(), false, false, false);
iterator.remove();
}
listener.removePendingConfirmsReference(this, entry.getValue());
}
@@ -706,7 +707,7 @@ public class PublisherCallbackChannelImpl
if (logger.isDebugEnabled()) {
logger.debug(this.toString() + " PC:Ack:" + seq + ":" + multiple);
}
this.processAck(seq, true, multiple);
this.processAck(seq, true, multiple, true);
}
public void handleNack(long seq, boolean multiple)
@@ -714,10 +715,10 @@ public class PublisherCallbackChannelImpl
if (logger.isDebugEnabled()) {
logger.debug(this.toString() + " PC:Nack:" + seq + ":" + multiple);
}
this.processAck(seq, false, multiple);
this.processAck(seq, false, multiple, true);
}
private synchronized void processAck(long seq, boolean ack, boolean multiple) {
private synchronized void processAck(long seq, boolean ack, boolean multiple, boolean remove) {
if (multiple) {
/*
* Piggy-backed ack - extract all Listeners for this and earlier
@@ -750,7 +751,13 @@ public class PublisherCallbackChannelImpl
Listener listener = this.listenerForSeq.remove(seq);
if (listener != null) {
SortedMap<Long, PendingConfirm> confirmsForListener = this.pendingConfirms.get(listener);
PendingConfirm pendingConfirm = confirmsForListener.remove(seq);
PendingConfirm pendingConfirm;
if (remove) {
pendingConfirm = confirmsForListener.remove(seq);
}
else {
pendingConfirm = confirmsForListener.get(seq);
}
if (pendingConfirm != null) {
doHandleConfirm(ack, listener, pendingConfirm);
}

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

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -64,6 +65,8 @@ import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.PendingConfirm;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannel.Listener;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
@@ -81,6 +84,7 @@ import com.rabbitmq.client.ConnectionFactory;
* @author Gary Russell
* @author Gunar Hillert
* @author Artem Bilan
* @author Rolf Arne Corneliussen
* @since 1.1
*
*/
@@ -892,4 +896,44 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertTrue(confirmed.get());
}
@Test
public void testPublisherCallbackChannelImplCloseWithPending() throws Exception {
final AtomicInteger nacks = new AtomicInteger();
Listener listener = mock(Listener.class);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
boolean ack = (Boolean) invocation.getArguments()[1];
if (!ack) {
nacks.incrementAndGet();
}
return null;
}
}).when(listener).handleConfirm(any(PendingConfirm.class), anyBoolean());
when(listener.getUUID()).thenReturn(UUID.randomUUID().toString());
when(listener.isConfirmListener()).thenReturn(true);
Channel channelMock = mock(Channel.class);
PublisherCallbackChannelImpl channel = new PublisherCallbackChannelImpl(channelMock);
channel.addListener(listener);
for (int i = 0; i < 2; i++) {
long seq = i + 1000;
channel.addPendingConfirm(listener, seq,
new PendingConfirm(new CorrelationData(Long.toHexString(seq)), System.currentTimeMillis()));
}
channel.close();
assertEquals(0, TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size());
assertEquals(2, nacks.get());
}
}