diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 50457975..1a2d04b5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -691,6 +691,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory } try { if (this.target == null || !this.target.isOpen()) { + if (this.target instanceof PublisherCallbackChannel) { + this.target.close(); + throw new InvocationTargetException(new AmqpException("PublisherCallbackChannel is closed")); + } this.target = null; } synchronized (targetMonitor) { @@ -703,10 +707,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory catch (InvocationTargetException ex) { if (this.target == null || !this.target.isOpen()) { // Basic re-connection logic... - this.target = null; if (logger.isDebugEnabled()) { - logger.debug("Detected closed channel on exception. Re-initializing: " + target); + logger.debug("Detected closed channel on exception. Re-initializing: " + this.target); } + this.target = null; synchronized (targetMonitor) { if (this.target == null) { this.target = createBareChannel(theConnection, transactional); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 4a248a39..bcf83adb 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -548,7 +548,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, long threshold = System.currentTimeMillis() - age; for (Entry> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) { SortedMap channelPendingConfirms = channelPendingConfirmEntry.getValue(); - synchronized(channelPendingConfirms) { + synchronized(channelPendingConfirmEntry.getKey()) { // channel Iterator> iterator = channelPendingConfirms.entrySet().iterator(); PendingConfirm pendingConfirm; while (iterator.hasNext()) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java index 3207bfc4..e33ac385 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -635,29 +635,37 @@ public class PublisherCallbackChannelImpl generateNacksForPendingAcks("Channel closed by application"); } - private void generateNacksForPendingAcks(String cause) { - synchronized (this.pendingConfirms) { - for (Entry> entry : this.pendingConfirms.entrySet()) { - Listener listener = entry.getKey(); - synchronized(entry.getValue()) { - for (Entry confirmEntry : entry.getValue().entrySet()) { - try { - confirmEntry.getValue().setCause(cause); - handleNack(confirmEntry.getKey(), false); - } - catch (IOException e) { - logger.error("Error delivering Nack afterShutdown", e); - } - } - listener.removePendingConfirmsReference(this, entry.getValue()); + private synchronized void generateNacksForPendingAcks(String cause) { + for (Entry> entry : this.pendingConfirms.entrySet()) { + Listener listener = entry.getKey(); + for (Entry confirmEntry : entry.getValue().entrySet()) { + try { + confirmEntry.getValue().setCause(cause); + handleNack(confirmEntry.getKey(), false); + } + catch (IOException e) { + logger.error("Error delivering Nack afterShutdown", e); } } - this.pendingConfirms.clear(); - this.listenerForSeq.clear(); - this.listeners.clear(); + listener.removePendingConfirmsReference(this, entry.getValue()); } + if (logger.isDebugEnabled()) { + logger.debug("PendingConfirms cleared"); + } + this.pendingConfirms.clear(); + this.listenerForSeq.clear(); + this.listeners.clear(); } + /** + * Add the listener and return the internal map of pending confirmations for that listener. + * Callers must synchronize on this channel object when modifying the map. + * This method will be changed in a future release to NOT expose the map. + * @param listener the listener. + * @return the internal map of pending confirmations. + * TODO: do not expose the map externally; change the {@code RabbitTemplate#getUnconfirmed(long)} + * functionality to delegate to a method here. + */ public synchronized SortedMap addListener(Listener listener) { Assert.notNull(listener, "Listener cannot be null"); if (this.listeners.size() == 0) { @@ -689,6 +697,9 @@ public class PublisherCallbackChannelImpl } } this.pendingConfirms.remove(listener); + if (logger.isDebugEnabled()) { + logger.debug("Removed listener " + listener); + } return result; } @@ -711,47 +722,40 @@ public class PublisherCallbackChannelImpl this.processAck(seq, false, multiple); } - private void processAck(long seq, boolean ack, boolean multiple) { + private synchronized void processAck(long seq, boolean ack, boolean multiple) { if (multiple) { /* * Piggy-backed ack - extract all Listeners for this and earlier * sequences. Then, for each Listener, handle each of it's acks. * Finally, remove the sequences from listenerForSeq. */ - synchronized(this.pendingConfirms) { - Map involvedListeners = this.listenerForSeq.headMap(seq + 1); - // eliminate duplicates - Set listeners = new HashSet(involvedListeners.values()); - for (Listener involvedListener : listeners) { - // find all unack'd confirms for this listener and handle them - SortedMap confirmsMap = this.pendingConfirms.get(involvedListener); - if (confirmsMap != null) { - synchronized(confirmsMap) { - Map confirms = confirmsMap.headMap(seq + 1); - Iterator> iterator = confirms.entrySet().iterator(); - while (iterator.hasNext()) { - Entry entry = iterator.next(); - PendingConfirm value = entry.getValue(); - iterator.remove(); - doHandleConfirm(ack, involvedListener, value); - } - } + Map involvedListeners = this.listenerForSeq.headMap(seq + 1); + // eliminate duplicates + Set listeners = new HashSet(involvedListeners.values()); + for (Listener involvedListener : listeners) { + // find all unack'd confirms for this listener and handle them + SortedMap confirmsMap = this.pendingConfirms.get(involvedListener); + if (confirmsMap != null) { + Map confirms = confirmsMap.headMap(seq + 1); + Iterator> iterator = confirms.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + PendingConfirm value = entry.getValue(); + iterator.remove(); + doHandleConfirm(ack, involvedListener, value); } } - List seqs = new ArrayList(involvedListeners.keySet()); - for (Long key : seqs) { - this.listenerForSeq.remove(key); - } + } + List seqs = new ArrayList(involvedListeners.keySet()); + for (Long key : seqs) { + this.listenerForSeq.remove(key); } } else { Listener listener = this.listenerForSeq.remove(seq); if (listener != null) { SortedMap confirmsForListener = this.pendingConfirms.get(listener); - PendingConfirm pendingConfirm = null; - synchronized (confirmsForListener) { - pendingConfirm = confirmsForListener.remove(seq); - } + PendingConfirm pendingConfirm = confirmsForListener.remove(seq); if (pendingConfirm != null) { doHandleConfirm(ack, listener, pendingConfirm); } @@ -776,12 +780,11 @@ public class PublisherCallbackChannelImpl } } - public void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { + public synchronized void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); - Assert.notNull(pendingConfirmsForListener, "Listener not registered"); - synchronized (pendingConfirmsForListener) { - pendingConfirmsForListener.put(seq, pendingConfirm); - } + Assert.notNull(pendingConfirmsForListener, + "Listener not registered: " + listener + " " + this.pendingConfirms.keySet()); + pendingConfirmsForListener.put(seq, pendingConfirm); this.listenerForSeq.put(seq, listener); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 7d648676..e5d01a7e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -56,6 +56,7 @@ import org.junit.Test; 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.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -342,6 +343,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -454,6 +456,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -497,6 +500,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -508,7 +512,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Override public Object answer(InvocationOnMock invocation) throws Throwable { return count.incrementAndGet(); - }}).when(mockChannel).getNextPublishSeqNo(); + } + }).when(mockChannel).getNextPublishSeqNo(); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setPublisherConfirms(true); @@ -542,6 +547,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -613,6 +619,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockChannel.getNextPublishSeqNo()).thenReturn(1L, 2L, 3L, 4L); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); @@ -841,14 +848,83 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { sentAll.set(true); } }); - Collection unconfirmed = template.getUnconfirmed(-1); long t1 = System.currentTimeMillis(); while (!sentAll.get() && System.currentTimeMillis() < t1 + 20000) { - unconfirmed = template.getUnconfirmed(-1); + template.getUnconfirmed(-1); } assertTrue(sentAll.get()); assertFalse(confirmed.get()); } + // AMQP-532 ConcurrentModificationException + @Test + public void testPublisherConfirmCloseConcurrency() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel1 = mock(Channel.class); + final AtomicLong seq1 = new AtomicLong(); + doAnswer(new Answer() { + + @Override + public Long answer(InvocationOnMock invocation) throws Throwable { + return seq1.incrementAndGet(); + } + }).when(mockChannel1).getNextPublishSeqNo(); + + Channel mockChannel2 = mock(Channel.class); + when(mockChannel2.isOpen()).thenReturn(true); + final AtomicLong seq2 = new AtomicLong(); + doAnswer(new Answer() { + + @Override + public Long answer(InvocationOnMock invocation) throws Throwable { + return seq2.incrementAndGet(); + } + }).when(mockChannel2).getNextPublishSeqNo(); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel1, mockChannel2); + + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); + ccf.setPublisherConfirms(true); + 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); + } + }); + ExecutorService exec = Executors.newSingleThreadExecutor(); + final AtomicInteger sent = new AtomicInteger(); + doAnswer(new Answer(){ + + @Override + public Boolean answer(InvocationOnMock invocation) throws Throwable { + boolean closed = sent.incrementAndGet() < 100; + System.out.println(closed); + return closed; + } + }).when(mockChannel1).isOpen(); + final CountDownLatch sentAll = new CountDownLatch(1); + exec.execute(new Runnable() { + + @Override + public void run() { + for (int i = 0; i < 1000; i++) { + try { + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + } + catch (AmqpException e) {} + } + sentAll.countDown(); + } + }); + assertTrue(sentAll.await(10, TimeUnit.SECONDS)); + assertTrue(confirmed.get()); + } }