GH-2152 Add support for honoring timeout for cases where destination may not yet exist

Resolves #2152

polishing

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,12 +17,15 @@
 package org.springframework.cloud.stream.binder.test;
 
 import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.Map;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.channel.AbstractSubscribableChannel;
 import org.springframework.messaging.Message;
 import org.springframework.util.StringUtils;
 
@@ -36,12 +39,14 @@ import org.springframework.util.StringUtils;
  */
 public class OutputDestination extends AbstractDestination {
 
-	private final Map<String, BlockingQueue<Message<byte[]>>> messageQueues = new LinkedHashMap<>();
+	private final Log log = LogFactory.getLog(OutputDestination.class);
+
+	private final ConcurrentHashMap<String, BlockingQueue<Message<byte[]>>> messageQueues = new ConcurrentHashMap<>();
 
 	public Message<byte[]> receive(long timeout, String bindingName) {
 		try {
 			bindingName = bindingName.endsWith(".destination") ? bindingName : bindingName + ".destination";
-			return this.messageQueues.get(bindingName).poll(timeout, TimeUnit.MILLISECONDS);
+			return this.outputQueue(bindingName).poll(timeout, TimeUnit.MILLISECONDS);
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
@@ -81,11 +86,13 @@ public class OutputDestination extends AbstractDestination {
 	 */
 	@Deprecated
 	public Message<byte[]> receive(long timeout, int bindingIndex) {
+		log.warn("!!!While 'receive(long timeout, int bindingIndex)' method may still work it is deprecated no longer supported. "
+				+ "It will be removed after 3.1.3 release. Please use 'receive(long timeout, String bindingName)'");
 		try {
 			BlockingQueue<Message<byte[]>> destinationQueue = (new ArrayList<>(this.messageQueues.values())).get(bindingIndex);
 			return destinationQueue.poll(timeout, TimeUnit.MILLISECONDS);
 		}
-		catch (InterruptedException e) {
+		catch (Exception e) {
 			Thread.currentThread().interrupt();
 		}
 		return null;
@@ -106,11 +113,13 @@ public class OutputDestination extends AbstractDestination {
 	@SuppressWarnings("unchecked")
 	@Override
 	void afterChannelIsSet(int channelIndex, String bindingName) {
-		if (!this.messageQueues.containsKey(bindingName)) {
+		if (((AbstractSubscribableChannel) this.getChannelByName(bindingName)).getSubscriberCount() < 1) {
-			BlockingQueue<Message<byte[]>> messageQueue = new LinkedTransferQueue<>();
+			this.getChannelByName(bindingName).subscribe(message -> this.outputQueue(bindingName).offer((Message<byte[]>) message));
-			this.messageQueues.put(bindingName, messageQueue);
-			this.getChannelByName(bindingName).subscribe(message -> this.messageQueues.get(bindingName).offer((Message<byte[]>) message));
 		}
 	}
 
+	private BlockingQueue<Message<byte[]>> outputQueue(String bindingName) {
+		this.messageQueues.putIfAbsent(bindingName, new LinkedTransferQueue<>());
+		return this.messageQueues.get(bindingName);
+	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -286,14 +286,14 @@ public class ImplicitFunctionBindingTests {
 			inputDestination.send(inputMessage);
 			inputDestination.send(inputMessage);
 			inputDestination.send(inputMessage);
-			assertThat(new String(outputDestination.receive(2000).getPayload())).isEqualTo("HelloHelloHello");
+			assertThat(new String(outputDestination.receive(2000, "aggregate-out-0").getPayload())).isEqualTo("HelloHelloHello");
-			assertThat(new String(outputDestination.receive(2000).getPayload())).isEqualTo("");
+			assertThat(new String(outputDestination.receive(2000, "aggregate-out-0").getPayload())).isEqualTo("");
 
 			inputDestination.send(inputMessage);
 			inputDestination.send(inputMessage);
 			inputDestination.send(inputMessage);
 			inputDestination.send(inputMessage);
-			assertThat(new String(outputDestination.receive(2000).getPayload())).isEqualTo("HelloHelloHelloHello");
+			assertThat(new String(outputDestination.receive(2000, "aggregate-out-0").getPayload())).isEqualTo("HelloHelloHelloHello");
 		}
 	}
 
@@ -565,16 +565,16 @@ public class ImplicitFunctionBindingTests {
 
 			OutputDestination outputDestination = context.getBean(OutputDestination.class);
 
-			Message<byte[]> outputMessage = outputDestination.receive(2000);
+			Message<byte[]> outputMessage = outputDestination.receive(2000, "supplierfunctionA-out-0");
 			Long value = Long.parseLong(new String(outputMessage.getPayload()));
 
-			outputMessage = outputDestination.receive(5000);
+			outputMessage = outputDestination.receive(5000, "supplierfunctionA-out-0");
 			assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000);
 
-			outputMessage = outputDestination.receive(5000);
+			outputMessage = outputDestination.receive(5000, "supplierfunctionA-out-0");
 			assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000);
 
-			outputMessage = outputDestination.receive(5000);
+			outputMessage = outputDestination.receive(5000, "supplierfunctionA-out-0");
 			assertThat(Long.parseLong(new String(outputMessage.getPayload())) - value).isGreaterThanOrEqualTo(1000);
 		}
 	}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java

@@ -240,40 +240,15 @@ public class SourceToFunctionsSupportTests {
 
 			OutputDestination target = context.getBean(OutputDestination.class);
 
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("1");
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("2");
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("3");
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("4");
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("5");
-			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6");
+			assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("6");
-
-			//assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null);
 		}
 	}
 
-	@Test
-	public void testMultipleSuppliers() {
-		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
-				TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
-						MultipleSupplierConfiguration.class)).web(WebApplicationType.NONE).run(
-								"--spring.cloud.function.definition=supplier1;supplier2",
-								"--spring.jmx.enabled=false"
-//								"--spring.cloud.stream.function.bindings.supplier1-out-0=output1",
-//								"--spring.cloud.stream.function.bindings.supplier2-out-0=output2"
-								)) {
-
-			OutputDestination target = context.getBean(OutputDestination.class);
-
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1");
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2");
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3");
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4");
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5");
-//			assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6");
-//
-//			assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null);
-		}
-	}
 
 	@EnableAutoConfiguration
 	public static class MessageFluxSupplierConfiguration {

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -57,6 +60,27 @@ public class StreamBridgeTests {
 		System.clearProperty("spring.cloud.function.definition");
 	}
 
+	@Test
+	public void testDelayedSend() {
+		ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(ConsumerConfiguration.class, EmptyConfiguration.class))
+						.web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false")) {
+
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			executor.schedule(() -> bridge.send("blah", "hello foo"), 5000, TimeUnit.MILLISECONDS);
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+			Message<byte[]> message = outputDestination.receive(10000, "blah");
+			assertThat(message).isNotNull();
+			assertThat(new String(message.getPayload())).isEqualTo("hello foo");
+		}
+		finally {
+			executor.shutdownNow();
+		}
+	}
+
 	@Test
 	public void testWithInterceptor() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration