From 211647d6fe7cc6e5893c2842fd6ddbce8f3f3d64 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 6 Oct 2016 09:44:19 -0400 Subject: [PATCH] INT-4129: Add Discard Channel to Barrier Handler JIRA: https://jira.spring.io/browse/INT-4129 Discard late arriving triggers. * Minor code style polishing Conflicts: src/reference/asciidoc/whats-new.adoc Resolved. * Fix trailing whitespaces in the `BarrierMessageHandlerTests` --- .../aggregator/BarrierMessageHandler.java | 44 ++++++++++++++++++- .../integration/config/xml/BarrierParser.java | 1 + .../config/spring-integration-4.3.xsd | 23 +++++++--- .../BarrierMessageHandlerTests.java | 40 ++++++++++++----- .../config/xml/BarrierParserTests-context.xml | 6 ++- .../config/xml/BarrierParserTests.java | 5 +++ src/reference/asciidoc/barrier.adoc | 3 ++ src/reference/asciidoc/whats-new.adoc | 5 +++ 8 files changed, 108 insertions(+), 19 deletions(-) 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 96d5464886..bfa833b756 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,6 +16,7 @@ package org.springframework.integration.aggregator; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.SynchronousQueue; @@ -23,9 +24,11 @@ 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,9 +49,10 @@ 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 = + private final Map>> suspensions = new ConcurrentHashMap>>(); private final ConcurrentMap inProcess = new ConcurrentHashMap(); @@ -59,6 +63,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 +114,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 4.3.5 + */ + 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 4.3.5 + */ + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + /** + * @since 4.3.5 + */ + @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"; @@ -186,6 +223,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-4.3.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd index 4e3e897b85..da1eb3ee99 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd @@ -1744,11 +1744,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 df3cfc5d85..5a526e5137 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; @@ -61,6 +63,8 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -126,7 +130,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 +148,15 @@ public class BarrierMessageHandlerTests { handler.setOutputChannel(outputChannel); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); - Executors.newSingleThreadExecutor().execute(new Runnable() { + Executors.newSingleThreadExecutor() + .execute(new Runnable() { - @Override - public void run() { - handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()); - } + @Override + public void run() { + 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,7 +176,17 @@ public class BarrierMessageHandlerTests { public void testLateReply() throws Exception { final BarrierMessageHandler handler = new BarrierMessageHandler(0); QueueChannel outputChannel = new QueueChannel(); + final QueueChannel discardChannel = new QueueChannel(); handler.setOutputChannel(outputChannel); + handler.setDiscardChannelName("discards"); + handler.setChannelResolver(new DestinationResolver() { + + @Override + public MessageChannel resolveDestination(String s) throws DestinationResolutionException { + return discardChannel; + } + + }); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); final CountDownLatch latch = new CountDownLatch(1); @@ -189,13 +204,16 @@ public class BarrierMessageHandlerTests { 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()); } @@ -300,7 +318,7 @@ public class BarrierMessageHandlerTests { return barrier; } - @ServiceActivator (inputChannel = "release", poller = @Poller(fixedDelay = "0")) + @ServiceActivator(inputChannel = "release", poller = @Poller(fixedDelay = "0")) @Bean public MessageHandler releaser() { return new MessageHandler() { @@ -315,6 +333,6 @@ public class BarrierMessageHandlerTests { }; } - } + } } 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 bd16ca8c7e..017b471025 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -290,3 +290,8 @@ See <> for more information. The `ServerWebSocketContainer` now exposes `allowedOrigins` option and `SockJsServiceOptions` a `suppressCors` option. 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.