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.

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();
 

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