Fixing TCP Source issues

Exclude `IpHeaders.LOCAL_ADDRESS` from TCP source output as it
causes issues with message conversion downstream.

Update SCSt/Spring-Cloud to the latest snapshots.

* Refactor `TcpSupplierConfiguration` for better readability and
current state of requirements for Spring libraries configuration

applications/stream-applications-core/pom.xml

@@ -21,8 +21,8 @@
         <java-functions.version>1.0.3-SNAPSHOT</java-functions.version>
         <apps.base-image>springcloud/baseimage:1.0.0</apps.base-image>
         <mockserver.version>5.10</mockserver.version>
-        <spring-cloud-stream-dependencies.version>Horsham.SR11</spring-cloud-stream-dependencies.version>
-        <spring-cloud-stream.version>3.0.11.RELEASE</spring-cloud-stream.version>
+        <spring-cloud-stream-dependencies.version>Horsham.BUILD-SNAPSHOT</spring-cloud-stream-dependencies.version>
+        <spring-cloud-stream.version>3.0.12.BUILD-SNAPSHOT</spring-cloud-stream.version>
         <spring-cloud-dataflow-apps-generator-plugin.version>1.0.2</spring-cloud-dataflow-apps-generator-plugin.version>
         <spring-cloud-dataflow-apps-docs-plugin.version>1.0.2</spring-cloud-dataflow-apps-docs-plugin.version>
         <spring-cloud-dataflow-apps-metadata-plugin.version>1.0.2</spring-cloud-dataflow-apps-metadata-plugin.version>
@@ -30,7 +30,7 @@
         <prometheus-rsocket.version>1.2.1</prometheus-rsocket.version>
         <!-- Boot 2.4.x needs wavefront-spring-boot version 2.1.0-RC1+ -->
         <wavefront-spring-boot.version>2.0.2</wavefront-spring-boot.version>
-        <spring-cloud.version>Hoxton.SR10</spring-cloud.version>
+        <spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
     </properties>
 
     <modules>

functions/supplier/tcp-supplier/README.adoc

@@ -2,15 +2,14 @@
 
 This module provides a TCP supplier that can be reused and composed in other applications.
 The `Supplier` uses the `TcpReceivingChannelAdapter` from Spring Integration.
-`tcpSupplier` is implemented as a `java.util.function.Supplier`.
-This supplier gives you a reactive stream from TCP sources. The supplier has a signature of `Supplier<Flux<Message<?>>>`.
+A `tcpSupplier` bean is implemented as a `java.util.function.Supplier`.
+This supplier gives you a reactive stream from TCP sources.
+The supplier has a signature of `Supplier<Flux<Message<?>>>`.
 Users have to subscribe to this `Flux` and then receive the data.
 
 ## Beans for injection
 
-You can import the `TcpSupplierConfiguration` in the application and then inject the following bean.
-
-`tcpSupplier`
+You can import the `TcpSupplierConfiguration` in the application and then inject the following bean: `tcpSupplier`.
 
 You need to inject this as `Supplier<Flux<Message<?>>>`.
 

functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-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.
@@ -18,16 +18,17 @@ package org.springframework.cloud.fn.supplier.tcp;
 
 import java.util.function.Supplier;
 
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.fn.common.tcp.EncoderDecoderFactoryBean;
 import org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
 import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
@@ -40,59 +41,59 @@ import org.springframework.messaging.Message;
  * @author Gary Russell
  * @author Christian Tzolov
  * @author Soby Chacko
+ * @author Artem Bilan
  */
-@Configuration
-@EnableConfigurationProperties({TcpSupplierProperties.class, TcpConnectionFactoryProperties.class})
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties({ TcpSupplierProperties.class, TcpConnectionFactoryProperties.class })
 public class TcpSupplierConfiguration {
 
-	@Autowired
-	private TcpSupplierProperties properties;
-
-	@Autowired
-	private TcpConnectionFactoryProperties tcpConnectionProperties;
-
-	@Qualifier("tcpSourceConnectionFactory")
-	@Autowired
-	private AbstractConnectionFactory connectionFactory;
-
 	@Bean
-	public Supplier<Flux<Message<?>>> tcpSupplier() {
-		return () -> Flux.from(output())
-				.doOnSubscribe(subscription -> adapter().start());
+	public EncoderDecoderFactoryBean tcpSourceDecoder(TcpSupplierProperties properties) {
+		EncoderDecoderFactoryBean factoryBean = new EncoderDecoderFactoryBean(properties.getDecoder());
+		factoryBean.setMaxMessageSize(properties.getBufferSize());
+		return factoryBean;
 	}
 
 	@Bean
-	public TcpReceivingChannelAdapter adapter() {
+	public TcpConnectionFactoryFactoryBean tcpSourceConnectionFactory(
+			TcpConnectionFactoryProperties tcpConnectionProperties,
+			@Qualifier("tcpSourceDecoder") AbstractByteArraySerializer decoder) {
+
+		TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
+		factoryBean.setType("server");
+		factoryBean.setPort(tcpConnectionProperties.getPort());
+		factoryBean.setUsingNio(tcpConnectionProperties.isNio());
+		factoryBean.setUsingDirectBuffers(tcpConnectionProperties.isUseDirectBuffers());
+		factoryBean.setLookupHost(tcpConnectionProperties.isReverseLookup());
+		factoryBean.setDeserializer(decoder);
+		factoryBean.setSoTimeout(tcpConnectionProperties.getSocketTimeout());
+		return factoryBean;
+	}
+
+	@Bean
+	public TcpReceivingChannelAdapter adapter(
+			@Qualifier("tcpSourceConnectionFactory") AbstractConnectionFactory connectionFactory) {
+
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(connectionFactory);
-		adapter.setOutputChannel(output());
 		adapter.setAutoStartup(false);
 		return adapter;
 	}
 
 	@Bean
-	public FluxMessageChannel output() {
-		return new FluxMessageChannel();
+	public Publisher<Message<Object>> tcpSupplierFlow(TcpReceivingChannelAdapter adapter) {
+		return IntegrationFlows.from(adapter)
+				.headerFilter(IpHeaders.LOCAL_ADDRESS)
+				.toReactivePublisher();
 	}
 
 	@Bean
-	public TcpConnectionFactoryFactoryBean tcpSourceConnectionFactory(
-			@Qualifier("tcpSourceDecoder") AbstractByteArraySerializer decoder) {
-		TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
-		factoryBean.setType("server");
-		factoryBean.setPort(this.tcpConnectionProperties.getPort());
-		factoryBean.setUsingNio(this.tcpConnectionProperties.isNio());
-		factoryBean.setUsingDirectBuffers(this.tcpConnectionProperties.isUseDirectBuffers());
-		factoryBean.setLookupHost(this.tcpConnectionProperties.isReverseLookup());
-		factoryBean.setDeserializer(decoder);
-		factoryBean.setSoTimeout(this.tcpConnectionProperties.getSocketTimeout());
-		return factoryBean;
+	public Supplier<Flux<Message<Object>>> tcpSupplier(
+			Publisher<Message<Object>> tcpSupplierFlow,
+			TcpReceivingChannelAdapter tcpReceivingChannelAdapter) {
+
+		return () -> Flux.from(tcpSupplierFlow)
+				.doOnSubscribe(subscription -> tcpReceivingChannelAdapter.start());
 	}
 
-	@Bean
-	public EncoderDecoderFactoryBean tcpSourceDecoder() {
-		EncoderDecoderFactoryBean factoryBean = new EncoderDecoderFactoryBean(this.properties.getDecoder());
-		factoryBean.setMaxMessageSize(this.properties.getBufferSize());
-		return factoryBean;
-	}
 }

functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-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.
@@ -38,9 +38,9 @@ import static org.assertj.core.api.Assertions.assertThat;
  *
  * @author Gary Russell
  * @author Soby Chacko
+ * @author Artem Bilan
  */
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
-		properties = { "tcp.host = localhost", "tcp.port = 0" })
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = "tcp.port = 0")
 @DirtiesContext
 public class AbstractTcpSupplierTests {