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`

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<Object, SynchronousQueue<Message<?>>> suspensions =
+	private final Map<Object, SynchronousQueue<Message<?>>> suspensions =
 			new ConcurrentHashMap<Object, SynchronousQueue<Message<?>>>();
 
 	private final ConcurrentMap<Object, Thread> inProcess = new ConcurrentHashMap<Object, Thread>();
@@ -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) {

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;
 	}
 

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd

@@ -1744,11 +1744,24 @@
 							<tool:expected-type type="org.springframework.integration.aggregator.MessageGroupProcessor" />
 						</tool:annotation>
 					</xsd:appinfo>
-						<xsd:documentation>
-							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.
-						</xsd:documentation>
+					<xsd:documentation>
+						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.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="discard-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						The message channel to which to send a trigger message if it arrives after the main
+						thread has timed out.
+					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
 		</xsd:complexType>

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<MessageChannel>() {
+
+			@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<String> triggerMessage = MessageBuilder.withPayload("bar").setCorrelationId("foo").build();
+		handler.trigger(triggerMessage);
 		ArgumentCaptor<String> 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 {
 			};
 		}
 
- 	}
+	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml

@@ -10,7 +10,7 @@
 	</int:channel>
 
 	<int:barrier id="barrier1" input-channel="in" output-channel="out" correlation-strategy-expression="'foo'"
-				requires-reply="true"
+				requires-reply="true" discard-channel="discards"
 				timeout="10000">
 		<int:poller fixed-delay="100" />
 	</int:barrier>
@@ -19,6 +19,10 @@
 		<int:queue />
 	</int:channel>
 
+	<int:channel id="discards">
+		<int:queue />
+	</int:channel>
+
 	<int:channel id="release" />
 
 	<int:outbound-channel-adapter channel="release" ref="barrier1.handler" method="trigger" />

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 {

src/reference/asciidoc/barrier.adoc

@@ -44,6 +44,7 @@ An exception is thrown if a second thread arrives with the same correlation.
 <int:barrier id="barrier1" input-channel="in" output-channel="out"
         correlation-strategy-expression="headers['myHeader']"
         output-processor="myOutputProcessor"
+        discard-channel="lateTriggerChannel"
         timeout="10000">
 </int:barrier>
 
@@ -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;
     }
 

src/reference/asciidoc/whats-new.adoc

@@ -290,3 +290,8 @@ See <<jdbc-message-store-channels>> for more information.
 
 The `ServerWebSocketContainer` now exposes `allowedOrigins` option and `SockJsServiceOptions` a `suppressCors` option.
 See <<web-sockets>> for more information.
+
+==== Barrier Changes
+
+The `BarrierMessageHandler` now supports a discard channel to which late-arriving trigger messages are sent.
+See <<barrier>> for more information.