diff --git a/applications/source/pom.xml b/applications/source/pom.xml index 342d20f4..a6ba358a 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -15,5 +15,6 @@ http-source time-source mongodb-source + tcp-source diff --git a/applications/source/tcp-source/README.adoc b/applications/source/tcp-source/README.adoc new file mode 100644 index 00000000..14cc15e9 --- /dev/null +++ b/applications/source/tcp-source/README.adoc @@ -0,0 +1,38 @@ +//tag::ref-doc[] += TCP +The `tcp` source acts as a server and allows a remote party to connect to it and submit data over a raw tcp socket. + +TCP is a streaming protocol and some mechanism is needed to frame messages on the wire. A number of decoders are +available, the default being 'CRLF' which is compatible with Telnet. + +Messages produced by the TCP source application have a `byte[]` payload. + +== Options + +//tag::configuration-properties[] +$$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.supplier.buffer-size$$:: $$The buffer size used when decoding messages; larger messages will be rejected.$$ *($$Integer$$, default: `$$2048$$`)* +$$tcp.supplier.decoder$$:: $$The decoder to use when receiving messages.$$ *($$Encoding$$, default: `$$$$`, possible values: `CRLF`,`LF`,`NULL`,`STXETX`,`RAW`,`L1`,`L2`,`L4`)* +$$tcp.use-direct-buffers$$:: $$Whether or not to use direct buffers.$$ *($$Boolean$$, default: `$$false$$`)* +//end::configuration-properties[] + +== Available Decoders + +.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/source/tcp-source/pom.xml b/applications/source/tcp-source/pom.xml new file mode 100644 index 00000000..f8dc1698 --- /dev/null +++ b/applications/source/tcp-source/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + tcp-source + 3.0.0-SNAPSHOT + tcp-source + tcp source 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-supplier + ${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 + source + ${project.version} + org.springframework.cloud.fn.supplier.tcp.TcpSupplierConfiguration.class + + + + org.springframework.cloud.fn + tcp-supplier + ${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/source/tcp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/tcp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..e9463175 --- /dev/null +++ b/applications/source/tcp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.tcp.TcpSupplierProperties,\ + org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties diff --git a/applications/source/tcp-source/src/test/java/org/springframework/cloud/stream/app/source/tcp/TcpSourceTests.java b/applications/source/tcp-source/src/test/java/org/springframework/cloud/stream/app/source/tcp/TcpSourceTests.java new file mode 100644 index 00000000..fbbeec0a --- /dev/null +++ b/applications/source/tcp-source/src/test/java/org/springframework/cloud/stream/app/source/tcp/TcpSourceTests.java @@ -0,0 +1,98 @@ +/* + * Copyright 2020-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.source.tcp; + +import java.net.Socket; + +import javax.net.SocketFactory; + +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.supplier.tcp.TcpSupplierConfiguration; +import org.springframework.cloud.stream.binder.test.OutputDestination; +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.connection.AbstractServerConnectionFactory; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +public class TcpSourceTests { + + @Test + public void testBasicTcpSourceWithBinder() throws Exception { + + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(TcpSourceTestApplication.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=tcpSupplier")) { + + AbstractServerConnectionFactory connectionFactory = context.getBean(AbstractServerConnectionFactory.class); + + doTest("", "foo", "\r\n", connectionFactory); + + OutputDestination target = context.getBean(OutputDestination.class); + Message sourceMessage = target.receive(10000); + String actual = new String(sourceMessage.getPayload()); + assertThat(actual).isEqualTo("foo"); + + sourceMessage = target.receive(10000); + actual = new String(sourceMessage.getPayload()); + assertThat(actual).isEqualTo("foo"); + } + } + + protected void doTest(String prefix, String payload, String suffix, + AbstractServerConnectionFactory connectionFactory) throws Exception { + int port = getPort(connectionFactory); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write((prefix + payload + suffix).getBytes()); + if (prefix.length() == 0 && suffix.length() == 0) { + socket.close(); // RAW - for the others, close AFTER the messages are decoded. + socket = SocketFactory.getDefault().createSocket("localhost", port); + } + + socket.getOutputStream().write((prefix + payload + suffix).getBytes()); + if (prefix.length() == 0 && suffix.length() == 0) { + socket.close(); // RAW - for the others, close AFTER the messages are decoded. + } + socket.close(); + + } + + private int getPort(AbstractServerConnectionFactory connectionFactory) throws Exception { + int n = 0; + while (n++ < 100 && !connectionFactory.isListening()) { + Thread.sleep(100); + } + assertThat(connectionFactory.isListening()).isTrue(); + int port = connectionFactory.getPort(); + assertThat(port > 0).isTrue(); + return port; + } + + + @SpringBootApplication + @Import(TcpSupplierConfiguration.class) + public static class TcpSourceTestApplication { + } +} diff --git a/functions/common/tcp-common/pom.xml b/functions/common/tcp-common/pom.xml new file mode 100644 index 00000000..338f6502 --- /dev/null +++ b/functions/common/tcp-common/pom.xml @@ -0,0 +1,37 @@ + + + 4.0.0 + tcp-common + 1.0.0-SNAPSHOT + tcp-common + tcp common + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-ip + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java new file mode 100644 index 00000000..7cfe763c --- /dev/null +++ b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java @@ -0,0 +1,108 @@ +/* + * 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.common.tcp; + +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArraySingleTerminatorSerializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; +import org.springframework.util.Assert; + +/** + * Factory bean for an encoder/decoder based on + * {@link Encoding}. + * + * @author Gary Russell + * @author Christian Tzolov + */ +public class EncoderDecoderFactoryBean extends AbstractFactoryBean + implements ApplicationEventPublisherAware { + + private final Encoding encoding; + + private ApplicationEventPublisher applicationEventPublisher; + + private Integer maxMessageSize; + + public EncoderDecoderFactoryBean(Encoding encoding) { + Assert.notNull(encoding, "'encoding' cannot be null"); + this.encoding = encoding; + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + /** + * The maximum message size allowed when decoding. + * @param maxMessageSize the maximum message size. + */ + public void setMaxMessageSize(int maxMessageSize) { + this.maxMessageSize = maxMessageSize; + } + + @Override + protected AbstractByteArraySerializer createInstance() throws Exception { + AbstractByteArraySerializer codec; + switch (this.encoding) { + case CRLF: + codec = new ByteArrayCrLfSerializer(); + break; + case LF: + codec = new ByteArrayLfSerializer(); + break; + case NULL: + codec = new ByteArraySingleTerminatorSerializer((byte) 0); + break; + case STXETX: + codec = new ByteArrayStxEtxSerializer(); + break; + case L1: + codec = new ByteArrayLengthHeaderSerializer(1); + break; + case L2: + codec = new ByteArrayLengthHeaderSerializer(2); + break; + case L4: + codec = new ByteArrayLengthHeaderSerializer(4); + break; + case RAW: + codec = new ByteArrayRawSerializer(); + break; + default: + throw new IllegalArgumentException("Invalid encoding: " + this.encoding); + } + codec.setApplicationEventPublisher(this.applicationEventPublisher); + if (this.maxMessageSize != null) { + codec.setMaxMessageSize(this.maxMessageSize); + } + return codec; + } + + @Override + public Class getObjectType() { + return AbstractByteArraySerializer.class; + } + +} diff --git a/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java new file mode 100644 index 00000000..d434ce54 --- /dev/null +++ b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java @@ -0,0 +1,56 @@ +/* + * Copyright 2016-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.common.tcp; + +/** + * @author Gary Russell + * @author Christian Tzolov + */ +public enum Encoding { + /** + * CRLF encoding. + */ + CRLF, + /** + * LF encoding. + */ + LF, + /** + * Null encoding. + */ + NULL, + /** + * STXETX encoding. + */ + STXETX, + /** + * Raw encoding. + */ + RAW, + /** + * L1 encoding. + */ + L1, + /** + * L2 encoding. + */ + L2, + /** + * L4 encoding. + */ + L4; +} diff --git a/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java new file mode 100644 index 00000000..6d011dc5 --- /dev/null +++ b/functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java @@ -0,0 +1,96 @@ +/* + * 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.common.tcp; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Base class for TCP connection factory properties. + * + * @author Eric Bottard + * @author Gary Russell + * @author Christian Tzolov + */ +@ConfigurationProperties("tcp") +public class TcpConnectionFactoryProperties { + + /** + * The port on which to listen; 0 for the OS to choose a port. + */ + private int port = 1234; + + /** + * Perform a reverse DNS lookup on the remote IP Address; if false, + * just the IP address is included in the message headers. + */ + private boolean reverseLookup = false; + + /** + * The timeout (ms) before closing the socket when no data is received. + */ + private int socketTimeout = 120000; + + /** + * Whether or not to use NIO. + */ + private boolean nio = false; + + /** + * Whether or not to use direct buffers. + */ + private boolean useDirectBuffers = false; + + public int getPort() { + return this.port; + } + + public void setPort(int port) { + this.port = port; + } + + public boolean isUseDirectBuffers() { + return this.useDirectBuffers; + } + + public void setUseDirectBuffers(boolean useDirectBuffers) { + this.useDirectBuffers = useDirectBuffers; + } + + public boolean isNio() { + return this.nio; + } + + public void setNio(boolean nio) { + this.nio = nio; + } + + public int getSocketTimeout() { + return this.socketTimeout; + } + + public void setSocketTimeout(int socketTimeout) { + this.socketTimeout = socketTimeout; + } + + public boolean isReverseLookup() { + return this.reverseLookup; + } + + public void setReverseLookup(boolean reverseLookup) { + this.reverseLookup = reverseLookup; + } +} diff --git a/functions/pom.xml b/functions/pom.xml index c8f0dbd2..ed08a63d 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -42,6 +42,7 @@ common/ftp-common common/function-test-support + common/tcp-common consumer/cassandra-consumer consumer/counter-consumer @@ -64,6 +65,7 @@ supplier/http-supplier supplier/jdbc-supplier supplier/mongodb-supplier + supplier/tcp-supplier supplier/time-supplier spring-functions-parent diff --git a/functions/supplier/tcp-supplier/pom.xml b/functions/supplier/tcp-supplier/pom.xml new file mode 100644 index 00000000..0a9ebab9 --- /dev/null +++ b/functions/supplier/tcp-supplier/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + tcp-supplier + 1.0.0-SNAPSHOT + tcp-supplier + tcp supplier + + + 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/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java new file mode 100644 index 00000000..4730d862 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java @@ -0,0 +1,98 @@ +/* + * 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.supplier.tcp; + +import java.util.function.Supplier; + +import reactor.core.publisher.Flux; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.tcp.EncoderDecoderFactoryBean; +import org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; +import org.springframework.messaging.Message; + +/** + * A source module that receives data over TCP. + * + * @author Gary Russell + * @author Christian Tzolov + * @author Soby Chacko + */ +@Configuration +@EnableConfigurationProperties({TcpSupplierProperties.class, TcpConnectionFactoryProperties.class}) +public class TcpSupplierConfiguration { + + @Autowired + private TcpSupplierProperties properties; + + @Autowired + private TcpConnectionFactoryProperties tcpConnectionProperties; + + @Qualifier("tcpSourceConnectionFactory") + @Autowired + private AbstractConnectionFactory connectionFactory; + + @Bean + public Supplier>> tcpSupplier() { + return () -> Flux.from(output()) + .doOnSubscribe(subscription -> adapter().start()); + } + + @Bean + public TcpReceivingChannelAdapter adapter() { + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(connectionFactory); + adapter.setOutputChannel(output()); + adapter.setAutoStartup(false); + return adapter; + } + + @Bean + public FluxMessageChannel output() { + return new FluxMessageChannel(); + } + + @Bean + public TcpConnectionFactoryFactoryBean tcpSourceConnectionFactory( + @Qualifier("tcpSourceDecoder") AbstractByteArraySerializer decoder) { + TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean(); + factoryBean.setType("server"); + factoryBean.setPort(this.tcpConnectionProperties.getPort()); + factoryBean.setUsingNio(this.tcpConnectionProperties.isNio()); + factoryBean.setUsingDirectBuffers(this.tcpConnectionProperties.isUseDirectBuffers()); + factoryBean.setLookupHost(this.tcpConnectionProperties.isReverseLookup()); + factoryBean.setDeserializer(decoder); + factoryBean.setSoTimeout(this.tcpConnectionProperties.getSocketTimeout()); + return factoryBean; + } + + @Bean + public EncoderDecoderFactoryBean tcpSourceDecoder() { + EncoderDecoderFactoryBean factoryBean = new EncoderDecoderFactoryBean(this.properties.getDecoder()); + factoryBean.setMaxMessageSize(this.properties.getBufferSize()); + return factoryBean; + } +} diff --git a/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java new file mode 100644 index 00000000..6e403a30 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java @@ -0,0 +1,63 @@ +/* + * 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.supplier.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 Source. + * + * @author Gary Russell + * @author Christian Tzolov + * + */ +@ConfigurationProperties("tcp.supplier") +@Validated +public class TcpSupplierProperties { + + /** + * The decoder to use when receiving messages. + */ + private Encoding decoder = Encoding.CRLF; + + /** + * The buffer size used when decoding messages; larger messages will be rejected. + */ + private int bufferSize = 2048; + + @NotNull + public Encoding getDecoder() { + return this.decoder; + } + + public void setDecoder(Encoding decoder) { + this.decoder = decoder; + } + + public int getBufferSize() { + return bufferSize; + } + + public void setBufferSize(int bufferSize) { + this.bufferSize = bufferSize; + } + +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java new file mode 100644 index 00000000..8597a46d --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java @@ -0,0 +1,111 @@ +/* + * 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.supplier.tcp; + +import java.net.Socket; +import java.util.function.Supplier; + +import javax.net.SocketFactory; + +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +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.AbstractServerConnectionFactory; +import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for TCP Supplier. + * + * @author Gary Russell + * @author Soby Chacko + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { "tcp.host = localhost", "tcp.port = 0" }) +@DirtiesContext +public class AbstractTcpSupplierTests { + + @Autowired + Supplier>> tcpSupplier; + + @Autowired + protected AbstractServerConnectionFactory connectionFactory; + + @Autowired + protected TcpSupplierProperties properties; + + /* + * Sends two messages with and asserts the + * payload is received on the other side. + */ + protected void doTest(String prefix, String payload, String suffix) throws Exception { + + final Flux> messageFlux = tcpSupplier.get(); + + final StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext((message) -> { + assertThat(message.getPayload()) + .isEqualTo(payload.getBytes()); + } + ) + .assertNext((message) -> { + assertThat(message.getPayload()) + .isEqualTo(payload.getBytes()); + } + ) + .thenCancel() + .verifyLater(); + + + int port = getPort(); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write((prefix + payload + suffix).getBytes()); + if (prefix.length() == 0 && suffix.length() == 0) { + socket.close(); // RAW - for the others, close AFTER the messages are decoded. + socket = SocketFactory.getDefault().createSocket("localhost", port); + } + + socket.getOutputStream().write((prefix + payload + suffix).getBytes()); + if (prefix.length() == 0 && suffix.length() == 0) { + socket.close(); // RAW - for the others, close AFTER the messages are decoded. + } + socket.close(); + + stepVerifier.verify(); + } + + private int getPort() throws Exception { + int n = 0; + while (n++ < 100 && !this.connectionFactory.isListening()) { + Thread.sleep(100); + } + assertThat(this.connectionFactory.isListening()).isTrue(); + int port = this.connectionFactory.getPort(); + assertThat(port > 0).isTrue(); + return port; + } + + @SpringBootApplication + public static class TcpConsumerTestApplication { + + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/CRLFTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/CRLFTests.java new file mode 100644 index 00000000..a19543cb --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/CRLFTests.java @@ -0,0 +1,30 @@ +/* + * 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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +/** + * @author Gary Russell + */ +public class CRLFTests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("", "foo", "\r\n"); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L1Tests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L1Tests.java new file mode 100644 index 00000000..7aec2c43 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L1Tests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = L1" }) +public class L1Tests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("\u0003", "foo", ""); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L2Tests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L2Tests.java new file mode 100644 index 00000000..f2d7144c --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L2Tests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = L2" }) +public class L2Tests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("\u0000\u0003", "foo", ""); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L4Tests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L4Tests.java new file mode 100644 index 00000000..1d0210a1 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L4Tests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = L4" }) +public class L4Tests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("\u0000\u0000\u0000\u0003", "foo", ""); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/LFTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/LFTests.java new file mode 100644 index 00000000..cbb89888 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/LFTests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = LF" }) +public class LFTests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("", "foo", "\n"); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NULLTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NULLTests.java new file mode 100644 index 00000000..9aac7d4a --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NULLTests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = NULL" }) +public class NULLTests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("", "foo", "\u0000"); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NotNioTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NotNioTests.java new file mode 100644 index 00000000..2d7e039d --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NotNioTests.java @@ -0,0 +1,38 @@ +/* + * 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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.test.util.TestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Gary Russell + */ +public class NotNioTests extends AbstractTcpSupplierTests { + + @Test + public void test() { + assertThat(this.connectionFactory).isInstanceOf(TcpNetServerConnectionFactory.class); + assertThat(TestUtils.getPropertyValue(this.connectionFactory, "lookupHost", Boolean.class)).isFalse(); + assertThat(TestUtils.getPropertyValue(this.connectionFactory, "soTimeout")).isEqualTo(120000); + assertThat(TestUtils.getPropertyValue(this.connectionFactory, "deserializer.maxMessageSize")).isEqualTo(2048); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/PropertiesPopulatedTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/PropertiesPopulatedTests.java new file mode 100644 index 00000000..d3aff90a --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/PropertiesPopulatedTests.java @@ -0,0 +1,43 @@ +/* + * 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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; +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.nio = true", "tcp.reverseLookup = true", + "tcp.useDirectBuffers = true", "tcp.socketTimeout = 123", "tcp.supplier.bufferSize = 5"}) +public class PropertiesPopulatedTests extends AbstractTcpSupplierTests { + + @Test + public void test() { + assertThat(this.connectionFactory).isInstanceOf(TcpNioServerConnectionFactory.class); + 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(TestUtils.getPropertyValue(this.connectionFactory, "deserializer.maxMessageSize")).isEqualTo(5); + } + +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/RAWTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/RAWTests.java new file mode 100644 index 00000000..6ab80d62 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/RAWTests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = RAW" }) +public class RAWTests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("", "foo", ""); + } +} diff --git a/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/STXETXTests.java b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/STXETXTests.java new file mode 100644 index 00000000..1f3111a3 --- /dev/null +++ b/functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/STXETXTests.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.supplier.tcp; + +import org.junit.jupiter.api.Test; + +import org.springframework.test.context.TestPropertySource; + +/** + * @author Gary Russell + */ +@TestPropertySource(properties = { "tcp.supplier.decoder = STXETX" }) +public class STXETXTests extends AbstractTcpSupplierTests { + + @Test + public void test() throws Exception { + doTest("\u0002", "foo", "\u0003"); + } +}