TCP Sink

Migrating TCP Sink as a consumer/sink.

consumer/tcp-consumer/pom.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>tcp-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>tcp-consumer</name>
+    <description>tcp consumer</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>tcp-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import java.util.function.Consumer;
+
+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.SmartLifecycle;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
+import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
+import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
+import org.springframework.messaging.Message;
+
+/**
+ * A consumer that sends data over TCP.
+ *
+ * @author Gary Russell
+ * @author Christian Tzolov
+ */
+@Configuration
+@EnableConfigurationProperties({TcpConsumerProperties.class, TcpConnectionFactoryProperties.class})
+public class TcpConsumerConfiguration {
+
+	@Autowired
+	private TcpConsumerProperties properties;
+
+	@Autowired
+	private TcpConnectionFactoryProperties tcpConnectionProperties;
+
+	@Qualifier("tcpSinkConnectionFactory")
+	@Autowired
+	private AbstractConnectionFactory connectionFactory;
+
+	@Bean
+	public Consumer<Message<?>> tcpConsumer() {
+		return handler()::handleMessage;
+	}
+
+	@Bean
+	public TcpSendingMessageHandlerSmartLifeCycle handler() {
+		TcpSendingMessageHandlerSmartLifeCycle tcpMessageHandler = new TcpSendingMessageHandlerSmartLifeCycle();
+		tcpMessageHandler.setConnectionFactory(connectionFactory);
+		return tcpMessageHandler;
+	}
+
+	@Bean
+	public TcpConnectionFactoryFactoryBean tcpSinkConnectionFactory(
+			@Qualifier("tcpSinkEncoder") AbstractByteArraySerializer encoder,
+			@Qualifier("tcpSinkMapper") TcpMessageMapper mapper) throws Exception {
+		TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
+		factoryBean.setType("client");
+		factoryBean.setHost(this.properties.getHost());
+		factoryBean.setPort(this.tcpConnectionProperties.getPort());
+		factoryBean.setUsingNio(this.tcpConnectionProperties.isNio());
+		factoryBean.setUsingDirectBuffers(this.tcpConnectionProperties.isUseDirectBuffers());
+		factoryBean.setLookupHost(this.tcpConnectionProperties.isReverseLookup());
+		factoryBean.setSerializer(encoder);
+		factoryBean.setSoTimeout(this.tcpConnectionProperties.getSocketTimeout());
+		factoryBean.setMapper(mapper);
+		factoryBean.setSingleUse(this.properties.isClose());
+		return factoryBean;
+	}
+
+	@Bean
+	public EncoderDecoderFactoryBean tcpSinkEncoder() {
+		return new EncoderDecoderFactoryBean(this.properties.getEncoder());
+	}
+
+	@Bean
+	public TcpMessageMapper tcpSinkMapper() {
+		TcpMessageMapper mapper = new TcpMessageMapper();
+		mapper.setCharset(this.properties.getCharset());
+		return mapper;
+	}
+
+	static class TcpSendingMessageHandlerSmartLifeCycle extends TcpSendingMessageHandler implements SmartLifecycle {
+
+		@Override
+		public boolean isAutoStartup() {
+			return true;
+		}
+
+		@Override
+		public int getPhase() {
+			return Integer.MIN_VALUE;
+		}
+	}
+}

consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.fn.common.tcp.Encoding;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Properties for the TCP Consumer.
+ *
+ * @author Gary Russell
+ * @author Christian Tzolov
+ *
+ */
+@ConfigurationProperties("tcp.consumer")
+@Validated
+public class TcpConsumerProperties {
+
+	/**
+	 * The host to which this sink will connect.
+	 */
+	private String host;
+
+	/**
+	 * The encoder to use when sending messages.
+	 */
+	private Encoding encoder = Encoding.CRLF;
+
+	/**
+	 * The charset used when converting from bytes to String.
+	 */
+	private String charset = "UTF-8";
+
+	/**
+	 * Whether to close the socket after each message.
+	 */
+	private boolean close;
+
+	@NotNull
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	@NotNull
+	public Encoding getEncoder() {
+		return this.encoder;
+	}
+
+	public void setEncoder(Encoding encoder) {
+		this.encoder = encoder;
+	}
+
+	@NotNull
+	public String getCharset() {
+		return charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public boolean isClose() {
+		return close;
+	}
+
+	public void setClose(boolean close) {
+		this.close = close;
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import javax.net.ServerSocketFactory;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
+import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for TCP Consumer.
+ *
+ * @author Gary Russell
+ * @author Soby Chacko
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
+		properties = { "tcp.consumer.host = localhost", "tcp.port = ${tcp.consumer.test.port}" })
+@DirtiesContext
+public class AbstractTcpConsumerTests {
+
+	private static TestTCPServer server;
+
+	@Autowired
+	protected AbstractClientConnectionFactory connectionFactory;
+
+	@Autowired
+	Consumer<Message<?>> tcpConsumer;
+
+	@BeforeAll
+	public static void startup() {
+		server = new TestTCPServer();
+	}
+
+	@AfterAll
+	public static void shutDown() {
+		server.shutDown();
+	}
+
+
+	/*
+	 * Sends two messages and asserts they arrive as expected on the other side using
+	 * the supplied decoder.
+	 */
+	protected void doTest(AbstractByteArraySerializer decoder) throws Exception {
+		server.setDecoder(decoder);
+		Message<String> message = new GenericMessage<>("foo");
+		tcpConsumer.accept(message);
+		String received = server.queue.poll(10, TimeUnit.SECONDS);
+		assertThat(received).isEqualTo("foo");
+
+		tcpConsumer.accept(message);
+		received = server.queue.poll(10, TimeUnit.SECONDS);
+		assertThat(received).isEqualTo("foo");
+	}
+
+	/**
+	 * TCP server that uses the supplied {@link AbstractByteArraySerializer}
+	 * to decode the input stream and put the resulting message in a queue.
+	 *
+	 */
+	private static class TestTCPServer implements Runnable {
+
+		private static final Log logger = LogFactory.getLog(TestTCPServer.class);
+
+		private final ServerSocket serverSocket;
+
+		private final ExecutorService executor;
+
+		private volatile AbstractByteArraySerializer decoder;
+
+		private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
+
+		private volatile boolean stopped;
+
+		TestTCPServer() {
+			ServerSocket serverSocket = null;
+			ExecutorService executor = null;
+			try {
+				serverSocket = ServerSocketFactory.getDefault().createServerSocket(0);
+				System.setProperty("tcp.consumer.test.port", Integer.toString(serverSocket.getLocalPort()));
+				executor = Executors.newSingleThreadExecutor();
+			}
+			catch (IOException e) {
+				e.printStackTrace();
+			}
+			this.serverSocket = serverSocket;
+			this.executor = executor;
+			this.decoder = new ByteArrayCrLfSerializer();
+			executor.execute(this);
+		}
+
+		private void setDecoder(AbstractByteArraySerializer decoder) {
+			this.decoder = decoder;
+		}
+
+		@Override
+		public void run() {
+			while (true) {
+				Socket socket = null;
+				try {
+					logger.info("Server listening on " + this.serverSocket.getLocalPort());
+					socket = this.serverSocket.accept();
+					while (true) {
+						byte[] data = decoder.deserialize(socket.getInputStream());
+						queue.offer(new String(data));
+					}
+				}
+				catch (SoftEndOfStreamException e) {
+					// normal close
+				}
+				catch (IOException e) {
+					try {
+						if (socket != null) {
+							socket.close();
+						}
+					}
+					catch (IOException e1) {
+					}
+					logger.error(e.getMessage());
+					if (this.stopped) {
+						logger.info("Server stopped on " + this.serverSocket.getLocalPort());
+						break;
+					}
+				}
+			}
+		}
+
+		private void shutDown() {
+			try {
+				this.stopped = true;
+				this.serverSocket.close();
+				this.executor.shutdownNow();
+			}
+			catch (IOException e) {
+			}
+		}
+	}
+
+	@SpringBootApplication
+	public static class TcpConsumerTestApplication {
+
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+
+
+/**
+ * @author Gary Russell
+ */
+public class CRLFTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayCrLfSerializer());
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L1Tests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = L1" })
+public class L1Tests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayLengthHeaderSerializer(1));
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L2Tests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = L2" })
+public class L2Tests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayLengthHeaderSerializer(2));
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L4Tests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = L4" })
+public class L4Tests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayLengthHeaderSerializer(4));
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/LFTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = LF" })
+public class LFTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayLfSerializer());
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NULLTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArraySingleTerminatorSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = NULL" })
+public class NULLTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArraySingleTerminatorSerializer((byte) 0));
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NotNioTests.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.host = foo" })
+public class NotNioTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		assertThat(this.connectionFactory).isInstanceOf(TcpNetClientConnectionFactory.class);
+		assertThat(this.connectionFactory.getHost()).isEqualTo("foo");
+		assertThat(TestUtils.getPropertyValue(this.connectionFactory, "lookupHost", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(this.connectionFactory, "soTimeout")).isEqualTo(120000);
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/PropertiesPopulatedTests.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.host = foo", "tcp.nio = true", "tcp.reverseLookup = true",
+		"tcp.useDirectBuffers = true", "tcp.socketTimeout = 123", "tcp.consumer.close = true", "tcp.consumer.charset = bar" })
+public class PropertiesPopulatedTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		assertThat(this.connectionFactory).isInstanceOf(TcpNioClientConnectionFactory.class);
+		assertThat(this.connectionFactory.getHost()).isEqualTo("foo");
+		assertThat((TestUtils.getPropertyValue(this.connectionFactory, "lookupHost", Boolean.class))).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.connectionFactory, "usingDirectBuffers", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.connectionFactory, "soTimeout")).isEqualTo(123);
+		assertThat(this.connectionFactory.isSingleUse()).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.connectionFactory, "mapper.charset")).isEqualTo("bar");
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/RAWTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = RAW", "tcp.consumer.close = true" })
+public class RAWTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayRawSerializer());
+	}
+}

consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/STXETXTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.cloud.fn.consumer.tcp;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
+import org.springframework.test.context.TestPropertySource;
+
+/**
+ * @author Gary Russell
+ */
+@TestPropertySource(properties = { "tcp.consumer.encoder = STXETX" })
+public class STXETXTests extends AbstractTcpConsumerTests {
+
+	@Test
+	public void test() throws Exception {
+		doTest(new ByteArrayStxEtxSerializer());
+	}
+}

pom.xml

@@ -55,6 +55,7 @@
 		<module>consumer/rabbit-consumer</module>
 		<module>consumer/redis-consumer</module>
 		<module>consumer/sftp-consumer</module>
+		<module>consumer/tcp-consumer</module>
 
 		<module>function/filter-function</module>
 		<module>function/header-enricher-function</module>