From aaefe51909e903c2f89d2790eca1b4c6584e9d8f Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Wed, 24 Jul 2019 11:56:05 -0400
Subject: [PATCH] GH-3003: Fix pub/sub with dynamic DSL flows

Fixes https://github.com/spring-projects/spring-integration/issues/3003

Statically defined flows with a publish/subscribe channel invoke the
subscriptions in natural (declared) order.

The components in the flow are started by the application context in
phases (consumers, then producers) and bean declaration order within
each phase.

When a dynamically declared flow is started, the components are started
by the `StandardIntegrationFlow` in reverse order (last to first) so that
we don't start producing messages before the flow is fully wired.

This has the side-effect that pub/sub subscribers are invoked in an
unnatural (last to first) order.

All subscription sub-flows start with a bridge from the pub/sub channel
to the first component's input channel.

The `BroadcastingDispatcher` honors the `Ordered` interface.

Change the `PublishSubscribeSpec` to set the `order` property so that
subscribers are always invoked in the natural order, regardless of whether
the flow is statically or dynamically defined.
---
 .../integration/dsl/PublishSubscribeSpec.java |  5 ++++-
 .../PublishSubscribeTests.java                | 22 +++++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java
index f5ee5db071..9a654f26cd 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java
@@ -25,6 +25,7 @@ import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -32,6 +33,8 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
 
 	private final Map<Object, String> subscriberFlows = new LinkedHashMap<>();
 
+	private int order;
+
 	PublishSubscribeSpec() {
 		super();
 	}
@@ -50,7 +53,7 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
 
 		IntegrationFlowBuilder flowBuilder =
 				IntegrationFlows.from(this.channel)
-						.bridge();
+						.bridge(consumer -> consumer.order(this.order++));
 
 		MessageChannel subFlowInput = subFlow.getInputChannel();
 
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java
index a78458b862..1224912fc5 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java
@@ -31,6 +31,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.dsl.context.IntegrationFlowContext.IntegrationFlowRegistration;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.GenericMessage;
@@ -52,12 +54,28 @@ public class PublishSubscribeTests {
 	@Autowired
 	private List<Integer> subscribersOrderedCall;
 
+	@Autowired
+	private PubSubBugTestContext config;
+
+	@Autowired
+	private IntegrationFlowContext context;
+
 	@Test
 	public void executeFirstFlow() {
+		this.subscribersOrderedCall.clear();
 		this.inputChannel.send(new GenericMessage<>("Test"));
 		assertThat(this.subscribersOrderedCall).containsExactly(0, 1, 2, 3, 4, 5);
 	}
 
+	@Test
+	public void dynamicFlow() {
+		this.subscribersOrderedCall.clear();
+		IntegrationFlowRegistration reg = this.context.registration(this.config.flow()).register();
+		reg.getInputChannel().send(new GenericMessage<>("Test"));
+		assertThat(this.subscribersOrderedCall).containsExactly(0, 1, 2, 3, 4, 5);
+		this.context.remove(reg.getId());
+	}
+
 	@Configuration
 	@EnableIntegration
 	static class PubSubBugTestContext {
@@ -79,6 +97,10 @@ public class PublishSubscribeTests {
 
 		@Bean
 		public IntegrationFlow pubSubFlow() {
+			return flow();
+		}
+
+		IntegrationFlow flow() {
 			return f -> f
 					.publishSubscribeChannel(c -> c
 							.subscribe(sf -> sf