INT-4510: Test no memory leak in FluxMessageCh

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

Related to reactor/reactor-core#1290

The `Flux.publish()` and subsequent `connect()` doesn't fuse a subscriber
for the hooks flow is interrupted (complete, or error, or disconnect).
In this case the `FluxMessageChannel.publishers` store is not cleared
from the finished publishers

* Add test-case to check the `FluxMessageChannel.publishers` store after
finishing the stream
* Add `hide()` operator with TODO to remove when an appropriate Reactor
version is ready

**Cherry-pick to 5.0.x**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -83,6 +83,7 @@ public void subscribeTo(Publisher<Message<?>> publisher) {
 				Flux.from(publisher)
 						.doOnComplete(() -> this.publishers.remove(publisher))
 						.doOnNext(this::send)
+						.hide() // TODO remove after upgrade to Reactor 3.1.9.RELEASE or later
 						.publish();
 
 		this.publishers.put(publisher, connectableFlux);

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -26,8 +26,10 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -40,6 +42,11 @@
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.channel.MessageChannels;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.dsl.context.IntegrationFlowRegistration;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
@@ -53,6 +60,7 @@
 
 /**
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 @RunWith(SpringRunner.class)
@@ -68,8 +76,11 @@ public class FluxMessageChannelTests {
 	@Autowired
 	private PollableChannel errorChannel;
 
+	@Autowired
+	private IntegrationFlowContext integrationFlowContext;
+
 	@Test
-	public void testFluxMessageChannel() throws InterruptedException {
+	public void testFluxMessageChannel() {
 		QueueChannel replyChannel = new QueueChannel();
 
 		for (int i = 0; i < 10; i++) {
@@ -106,6 +117,31 @@ public void testMessageChannelReactiveAdaptation() throws InterruptedException {
 		assertThat(results, contains("FOO", "BAR"));
 	}
 
+	@Test
+	public void testFluxMessageChannelCleanUp() throws InterruptedException {
+		FluxMessageChannel flux = MessageChannels.flux().get();
+
+		CountDownLatch finishLatch = new CountDownLatch(1);
+
+		IntegrationFlow testFlow = f -> f
+				.<String>split(__ -> Flux.fromStream(IntStream.range(0, 100).boxed()), null)
+				.channel(flux)
+				.aggregate(a -> a.releaseStrategy(m -> m.size() == 100))
+				.handle(__ -> finishLatch.countDown());
+
+		IntegrationFlowRegistration flowRegistration =
+				this.integrationFlowContext.registration(testFlow)
+						.register();
+
+		flowRegistration.getInputChannel().send(new GenericMessage<>("foo"));
+
+		assertTrue(finishLatch.await(10, TimeUnit.SECONDS));
+
+		assertTrue(TestUtils.getPropertyValue(flux, "publishers", Map.class).isEmpty());
+
+		flowRegistration.destroy();
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class TestConfiguration {