From 3c047c5076450c8568cf0cab533754460573f454 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Tue, 25 Apr 2017 13:43:41 -0400
Subject: [PATCH] INT-4264: `.publishSubscribeChannel()`: fix order

JIRA: https://jira.spring.io/browse/INT-4264

When we use Java DSL `.publishSubscribeChannel()`, we expect that subscribers
will be called in the order they are defined in the flow.
But actually everything is depend on the populated `MessageHandler` implementation.
The `AbstractMessageHandler` implements `Ordered` and according the
`OrderedAwareCopyOnWriteArraySet` logic they are placed in the beginning of the
handlers set, meanwhile all others (un-ordered) are moved to the end.

* Add implicit `.bridge()` to all the subFlows added to the `PublishSubscribeSpec`.
This way all the subFlow subscribers will be ordered the same way (because of
`BridgeHandler`) and, therefore, will be preformed in the expected (as declared) order
---
 .../integration/dsl/PublishSubscribeSpec.java |   6 +-
 .../PublishSubscribeTests.java                | 108 ++++++++++++++++++
 2 files changed, 112 insertions(+), 2 deletions(-)
 create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java

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 0ee44d46bd..56bc88cfae 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -46,7 +46,9 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
 	}
 
 	public PublishSubscribeSpec subscribe(IntegrationFlow flow) {
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(this.channel);
+		IntegrationFlowBuilder flowBuilder =
+				IntegrationFlows.from(this.channel)
+						.bridge();
 		flow.configure(flowBuilder);
 		this.subscriberFlows.add(flowBuilder.get());
 		return _this();
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
new file mode 100644
index 0000000000..9e5f842904
--- /dev/null
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/publishsubscribe/PublishSubscribeTests.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.publishsubscribe;
+
+import static org.hamcrest.Matchers.contains;
+import static org.junit.Assert.assertThat;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.function.Consumer;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+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.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ * @author Gary Russell
+ *
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+public class PublishSubscribeTests {
+
+	@Autowired
+	@Qualifier("pubSubFlow.input")
+	private MessageChannel inputChannel;
+
+	@Autowired
+	private List<Integer> subscribersOrderedCall;
+
+	@Test
+	public void executeFirstFlow() {
+		this.inputChannel.send(new GenericMessage<>("Test"));
+		assertThat(this.subscribersOrderedCall, contains(0, 1, 2, 3, 4, 5));
+	}
+
+	@Configuration
+	@EnableIntegration
+	static class PubSubBugTestContext {
+
+		@Bean
+		public List<Integer> subscribersOrderedCall() {
+			return new LinkedList<>();
+		}
+
+		@Bean
+		public Consumer<String> subscriberConsumerBean() {
+			return s -> subscribersOrderedCall().add(2);
+		}
+
+		@Bean
+		public MessageHandler subscriberMessageHandlerBean() {
+			return s -> subscribersOrderedCall().add(3);
+		}
+
+		@Bean
+		public IntegrationFlow pubSubFlow() {
+			return f -> f
+					.publishSubscribeChannel(c -> c
+							.subscribe(sf -> sf
+									.handle(m -> subscribersOrderedCall().add(0)))
+							.subscribe(sf -> sf
+									.<String>handle((p, h) -> {
+										subscribersOrderedCall().add(1);
+										return null;
+									}))
+							.subscribe(sf -> sf
+									.handle(subscriberConsumerBean()))
+							.subscribe(sf -> sf
+									.handle(subscriberMessageHandlerBean()))
+							.subscribe(sf -> sf
+									.channel("secondInlineSubscriberChannel")
+									.handle(m -> subscribersOrderedCall().add(4)))
+					)
+					.<String>handle((p, h) -> {
+						subscribersOrderedCall().add(5);
+						return null;
+					});
+		}
+
+	}
+
+}