diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java index 3065c34ac2..e4143e09db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java @@ -16,16 +16,18 @@ package org.springframework.integration.aggregator; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.DiscardingMessageHandler; import org.springframework.integration.handler.MessageTriggerAction; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -46,12 +48,12 @@ import org.springframework.util.Assert; * * @since 4.2 */ -public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler implements MessageTriggerAction { +public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler + implements MessageTriggerAction, DiscardingMessageHandler { - private final ConcurrentMap>> suspensions = - new ConcurrentHashMap>>(); + private final Map>> suspensions = new ConcurrentHashMap<>(); - private final ConcurrentMap inProcess = new ConcurrentHashMap(); + private final Map inProcess = new ConcurrentHashMap<>(); private final long timeout; @@ -59,6 +61,10 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler private final MessageGroupProcessor messageGroupProcessor; + private volatile MessageChannel discardChannel; + + private String discardChannelName; + /** * Construct an instance with the provided timeout and default correlation and * output strategies. @@ -106,6 +112,35 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler this.timeout = timeout; } + /** + * Set the name of the channel to which late arriving trigger messages are sent. + * @param discardChannelName the discard channel. + * @since 5.0 + */ + public void setDiscardChannelName(String discardChannelName) { + this.discardChannelName = discardChannelName; + } + + /** + * Set the channel to which late arriving trigger messages are sent. + * @param discardChannel the discard channel. + * @since 5.0 + */ + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + /** + * @since 5.0 + */ + @Override + public MessageChannel getDiscardChannel() { + if (this.discardChannel == null && this.discardChannelName != null && getChannelResolver() != null) { + this.discardChannel = getChannelResolver().resolveDestination(this.discardChannelName); + } + return this.discardChannel; + } + @Override public String getComponentType() { return "barrier"; @@ -167,7 +202,7 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler } private SynchronousQueue> createOrObtainQueue(Object key) { - SynchronousQueue> syncQueue = new SynchronousQueue>(); + SynchronousQueue> syncQueue = new SynchronousQueue<>(); SynchronousQueue> existing = this.suspensions.putIfAbsent(key, syncQueue); if (existing != null) { syncQueue = existing; @@ -186,6 +221,9 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler if (!syncQueue.offer(message, this.timeout, TimeUnit.MILLISECONDS)) { this.logger.error("Suspending thread timed out or did not arrive within timeout for: " + message); this.suspensions.remove(key); + if (getDiscardChannel() != null) { + this.messagingTemplate.send(getDiscardChannel(), message); + } } } catch (InterruptedException e) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java index 29ef7ab219..64eb0bd03d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java @@ -45,6 +45,7 @@ public class BarrierParser extends AbstractConsumerEndpointParser { "correlation-strategy-method", "correlation-strategy-expression", "CorrelationStrategy", element, handlerBuilder, null, parserContext); IntegrationNamespaceUtils.setValueIfAttributeDefined(handlerBuilder, element, "requires-reply"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "discard-channel"); return handlerBuilder; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd index 7c821c0197..544156afd2 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd @@ -1731,11 +1731,24 @@ - - A reference to a bean that implements 'MessageGroupProcessor'. The processor is invoked to - produce the result when the release is triggered. By default the payloads of the two - messages are aggregated as a 'Collection' and the message headers are merged. - + + A reference to a bean that implements 'MessageGroupProcessor'. The processor is invoked to + produce the result when the release is triggered. By default the payloads of the two + messages are aggregated as a 'Collection' and the message headers are merged. + + + + + + + + + + + + The message channel to which to send a trigger message if it arrives after the main + thread has timed out. + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java index e1b8efaa41..57744e6a93 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java @@ -16,7 +16,9 @@ package org.springframework.integration.aggregator; +import static org.hamcrest.Matchers.allOf; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; @@ -99,19 +101,14 @@ public class BarrierMessageHandlerTests { handler.afterPropertiesSet(); final AtomicReference dupCorrelation = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - Runnable runnable = new Runnable() { - - @Override - public void run() { - try { - handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); - } - catch (MessagingException e) { - dupCorrelation.set(e); - } - latch.countDown(); + Runnable runnable = () -> { + try { + handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); } - + catch (MessagingException e) { + dupCorrelation.set(e); + } + latch.countDown(); }; ExecutorService exec = Executors.newCachedThreadPool(); exec.execute(runnable); @@ -126,7 +123,7 @@ public class BarrierMessageHandlerTests { assertTrue("suspension did not appear in time", n < 100); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertNotNull(dupCorrelation.get()); - assertThat(dupCorrelation.get().getMessage(), Matchers.startsWith("Correlation key (foo) is already in use by")); + assertThat(dupCorrelation.get().getMessage(), startsWith("Correlation key (foo) is already in use by")); handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()); Message received = outputChannel.receive(10000); assertNotNull(received); @@ -144,14 +141,8 @@ public class BarrierMessageHandlerTests { handler.setOutputChannel(outputChannel); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()); - } - - }); + Executors.newSingleThreadExecutor() + .execute(() -> handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build())); Map suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class); int n = 0; while (n++ < 100 && suspensions.size() == 0) { @@ -171,31 +162,32 @@ public class BarrierMessageHandlerTests { public void testLateReply() throws Exception { final BarrierMessageHandler handler = new BarrierMessageHandler(0); QueueChannel outputChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); handler.setOutputChannel(outputChannel); + handler.setDiscardChannelName("discards"); + handler.setChannelResolver(s -> discardChannel); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); final CountDownLatch latch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); - latch.countDown(); - } - + Executors.newSingleThreadExecutor().execute(() -> { + handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); + latch.countDown(); }); Map suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertEquals("suspension not removed", 0, suspensions.size()); Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class)); new DirectFieldAccessor(handler).setPropertyValue("logger", logger); - handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()); + final Message triggerMessage = MessageBuilder.withPayload("bar").setCorrelationId("foo").build(); + handler.trigger(triggerMessage); ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); verify(logger).error(captor.capture()); assertThat(captor.getValue(), - Matchers.allOf(containsString("Suspending thread timed out or did not arrive within timeout for:"), + allOf(containsString("Suspending thread timed out or did not arrive within timeout for:"), containsString("payload=bar"))); assertEquals(0, suspensions.size()); + Message discard = discardChannel.receive(0); + assertSame(discard, triggerMessage); handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); assertEquals(0, suspensions.size()); } @@ -226,19 +218,14 @@ public class BarrierMessageHandlerTests { handler.afterPropertiesSet(); final AtomicReference exception = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - try { - handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); - } - catch (Exception e) { - exception.set(e); - latch.countDown(); - } + Executors.newSingleThreadExecutor().execute(() -> { + try { + handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); + } + catch (Exception e) { + exception.set(e); + latch.countDown(); } - }); Map suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class); int n = 0; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml index b0e3e3273a..a07f48f87e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml @@ -10,7 +10,7 @@ @@ -19,6 +19,10 @@ + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java index ef4b6c1879..0749d33d27 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.config.xml; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -61,6 +62,9 @@ public class BarrierParserTests { @Autowired private PollableChannel out; + @Autowired + private PollableChannel discards; + @Autowired private PollingConsumer barrier1; @@ -91,6 +95,7 @@ public class BarrierParserTests { instanceOf(TestMGP.class)); assertThat(TestUtils.getPropertyValue(this.barrier3, "handler.correlationStrategy"), instanceOf(TestCS.class)); + assertSame(handler.getDiscardChannel(), this.discards); } public static class TestMGP implements MessageGroupProcessor { diff --git a/src/reference/asciidoc/barrier.adoc b/src/reference/asciidoc/barrier.adoc index f42cb95f4e..13736a903b 100644 --- a/src/reference/asciidoc/barrier.adoc +++ b/src/reference/asciidoc/barrier.adoc @@ -44,6 +44,7 @@ An exception is thrown if a second thread arrives with the same correlation. @@ -55,6 +56,7 @@ Either the thread sending a message to `in` or the one sending a message to `rel up to 10 seconds until the other arrives. When the message is released, the `out` channel will be sent a message combining the result of invoking the custom `MessageGroupProcessor` bean `myOutputProcessor`. +If the main thread times out and a trigger arrives later, you can configure a discard channel to which the late trigger will be sent. Java configuration is shown below. [source, java] @@ -68,6 +70,7 @@ public class Config { public BarrierMessageHandler barrier() { BarrierMessageHandler barrier = new BarrierMessageHandler(10000); barrier.setOutputChannel(out()); + barrier.setDiscardChannel(lateTriggers()); return barrier; } diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 738fed8ae6..ceb515428a 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -50,3 +50,8 @@ See <> for more information. There is a new option on the `CharacterStreamReadingMessageSource` to allow it to be used to "pipe" stdin and publish an application event when the pipe is closed. See <> for more information. + +==== Barrier Changes + +The `BarrierMessageHandler` now supports a discard channel to which late-arriving trigger messages are sent. +See <> for more information.