diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 203b3285..ac2b3edd 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -21,6 +21,7 @@ rabbit-sink router-sink sftp-sink + tcp-sink throughput-sink diff --git a/applications/sink/tcp-sink/README.adoc b/applications/sink/tcp-sink/README.adoc new file mode 100644 index 00000000..62aa0cf4 --- /dev/null +++ b/applications/sink/tcp-sink/README.adoc @@ -0,0 +1,41 @@ +//tag::ref-doc[] += TCP Sink + +This module writes messages to TCP using an Encoder. + +TCP is a streaming protocol and some mechanism is needed to frame messages on the wire. A number of encoders are +available, the default being 'CRLF'. + +== Options + +The **$$tcp$$** $$sink$$ has the following options: + +//tag::configuration-properties[] +$$tcp.consumer.charset$$:: $$The charset used when converting from bytes to String.$$ *($$String$$, default: `$$UTF-8$$`)* +$$tcp.consumer.close$$:: $$Whether to close the socket after each message.$$ *($$Boolean$$, default: `$$false$$`)* +$$tcp.consumer.encoder$$:: $$The encoder to use when sending messages.$$ *($$Encoding$$, default: `$$$$`, possible values: `CRLF`,`LF`,`NULL`,`STXETX`,`RAW`,`L1`,`L2`,`L4`)* +$$tcp.consumer.host$$:: $$The host to which this sink will connect.$$ *($$String$$, default: `$$$$`)* +$$tcp.nio$$:: $$Whether or not to use NIO.$$ *($$Boolean$$, default: `$$false$$`)* +$$tcp.port$$:: $$The port on which to listen; 0 for the OS to choose a port.$$ *($$Integer$$, default: `$$1234$$`)* +$$tcp.reverse-lookup$$:: $$Perform a reverse DNS lookup on the remote IP Address; if false, just the IP address is included in the message headers.$$ *($$Boolean$$, default: `$$false$$`)* +$$tcp.socket-timeout$$:: $$The timeout (ms) before closing the socket when no data is received.$$ *($$Integer$$, default: `$$120000$$`)* +$$tcp.use-direct-buffers$$:: $$Whether or not to use direct buffers.$$ *($$Boolean$$, default: `$$false$$`)* +//end::configuration-properties[] + +== Available Encoders + +.Text Data + +CRLF (default):: text terminated by carriage return (0x0d) followed by line feed (0x0a) +LF:: text terminated by line feed (0x0a) +NULL:: text terminated by a null byte (0x00) +STXETX:: text preceded by an STX (0x02) and terminated by an ETX (0x03) + +.Text and Binary Data + +RAW:: no structure - the client indicates a complete message by closing the socket +L1:: data preceded by a one byte (unsigned) length field (supports up to 255 bytes) +L2:: data preceded by a two byte (unsigned) length field (up to 2^16^-1 bytes) +L4:: data preceded by a four byte (signed) length field (up to 2^31^-1 bytes) + +//end::ref-doc[] diff --git a/applications/sink/tcp-sink/pom.xml b/applications/sink/tcp-sink/pom.xml new file mode 100644 index 00000000..e1b26011 --- /dev/null +++ b/applications/sink/tcp-sink/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + tcp-sink + 3.0.0-SNAPSHOT + tcp-sink + tcp sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + tcp-consumer + ${java-functions.version} + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + tcp + sink + ${project.version} + org.springframework.cloud.fn.consumer.tcp.TcpConsumerConfiguration.class + + + + org.springframework.cloud.fn + tcp-consumer + ${java-functions.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/tcp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/tcp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..80cea26d --- /dev/null +++ b/applications/sink/tcp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,3 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.tcp.TcpConsumerProperties,\ + org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties + diff --git a/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java b/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java new file mode 100644 index 00000000..f56eb900 --- /dev/null +++ b/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java @@ -0,0 +1,181 @@ +/* + * 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.stream.app.tcp.sink; + +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 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.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.fn.consumer.tcp.TcpConsumerConfiguration; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Import; +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.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +public class TcpSinkTests { + + private static TestTCPServer server; + + @BeforeAll + public static void setup() { + server = new TestTCPServer(); + } + + @AfterAll + public static void shutdown() { + server.shutDown(); + } + + @Test + public void testFileSink() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(TcpSinkTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=tcpConsumer", + "--tcp.consumer.host=localhost", + "--tcp.port=${tcp.consumer.test.port}")) { + + server.setDecoder(new ByteArrayCrLfSerializer()); + + Message message1 = MessageBuilder.withPayload("foo").build(); + InputDestination source = context.getBean(InputDestination.class); + source.send(message1); + + String received = server.queue.poll(10, TimeUnit.SECONDS); + assertThat(received).isEqualTo("foo"); + + Message message2 = MessageBuilder.withPayload("bar").build(); + source.send(message2); + + received = server.queue.poll(10, TimeUnit.SECONDS); + assertThat(received).isEqualTo("bar"); + } + } + + /** + * 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 + @Import(TcpConsumerConfiguration.class) + public static class TcpSinkTestApplication { + } +} diff --git a/functions/consumer/tcp-consumer/pom.xml b/functions/consumer/tcp-consumer/pom.xml new file mode 100644 index 00000000..a3bd42a1 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java b/functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java new file mode 100644 index 00000000..51195fae --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java b/functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java new file mode 100644 index 00000000..723ca43e --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java new file mode 100644 index 00000000..f1d92a92 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.java new file mode 100644 index 00000000..a59fc61d --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L1Tests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L1Tests.java new file mode 100644 index 00000000..3b20d27a --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L2Tests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L2Tests.java new file mode 100644 index 00000000..d7aa74c7 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L4Tests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/L4Tests.java new file mode 100644 index 00000000..275daff8 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/LFTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/LFTests.java new file mode 100644 index 00000000..2b6391c8 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NULLTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NULLTests.java new file mode 100644 index 00000000..d2e72f03 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NotNioTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/NotNioTests.java new file mode 100644 index 00000000..7dd9b6ef --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/PropertiesPopulatedTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/PropertiesPopulatedTests.java new file mode 100644 index 00000000..5fa219f5 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/RAWTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/RAWTests.java new file mode 100644 index 00000000..2087d544 --- /dev/null +++ b/functions/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/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/STXETXTests.java b/functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/STXETXTests.java new file mode 100644 index 00000000..25cdddc1 --- /dev/null +++ b/functions/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/functions/pom.xml b/functions/pom.xml index 360d1f90..44d8cb63 100644 --- a/functions/pom.xml +++ b/functions/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