diff --git a/consumer/tcp-consumer/pom.xml b/consumer/tcp-consumer/pom.xml new file mode 100644 index 00000000..a3bd42a1 --- /dev/null +++ b/consumer/tcp-consumer/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + tcp-consumer + 1.0.0-SNAPSHOT + tcp-consumer + tcp consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + tcp-common + ${project.version} + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.integration + spring-integration-test + test + + + + diff --git a/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java b/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java new file mode 100644 index 00000000..51195fae --- /dev/null +++ b/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> 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; + } + } +} diff --git a/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java b/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java new file mode 100644 index 00000000..723ca43e --- /dev/null +++ b/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; + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java new file mode 100644 index 00000000..f1d92a92 --- /dev/null +++ b/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> 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 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 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 { + + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.java new file mode 100644 index 00000000..a59fc61d --- /dev/null +++ b/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()); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L1Tests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L1Tests.java new file mode 100644 index 00000000..3b20d27a --- /dev/null +++ b/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)); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L2Tests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L2Tests.java new file mode 100644 index 00000000..d7aa74c7 --- /dev/null +++ b/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)); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L4Tests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L4Tests.java new file mode 100644 index 00000000..275daff8 --- /dev/null +++ b/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)); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/LFTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/LFTests.java new file mode 100644 index 00000000..2b6391c8 --- /dev/null +++ b/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()); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NULLTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NULLTests.java new file mode 100644 index 00000000..d2e72f03 --- /dev/null +++ b/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)); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NotNioTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NotNioTests.java new file mode 100644 index 00000000..7dd9b6ef --- /dev/null +++ b/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); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/PropertiesPopulatedTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/PropertiesPopulatedTests.java new file mode 100644 index 00000000..5fa219f5 --- /dev/null +++ b/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"); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/RAWTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/RAWTests.java new file mode 100644 index 00000000..2087d544 --- /dev/null +++ b/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()); + } +} diff --git a/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/STXETXTests.java b/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/STXETXTests.java new file mode 100644 index 00000000..25cdddc1 --- /dev/null +++ b/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()); + } +} diff --git a/pom.xml b/pom.xml index 360d1f90..44d8cb63 100644 --- a/pom.xml +++ b/pom.xml @@ -55,6 +55,7 @@ consumer/rabbit-consumer consumer/redis-consumer consumer/sftp-consumer + consumer/tcp-consumer function/filter-function function/header-enricher-function