SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

TCP Supplier/Source

Browse Source 
Resolves https://github.com/spring-cloud/stream-applications/issues/43
This commit is contained in:
Soby Chacko
2020-05-16 10:57:37 -04:00
committed by Gary Russell
parent 5073188c4c
commit 12fb2cfd46
24 changed files with 1182 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/source/pom.xml
View File

@@ -15,5 +15,6 @@
<module>http-source</module>
<module>time-source</module>
<module>mongodb-source</module>
<module>tcp-source</module>
</modules>
</project>

38
applications/source/tcp-source/README.adoc Normal file
View File

@@ -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: `$$<none>$$`, 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[]

78
applications/source/tcp-source/pom.xml Normal file
View File

@@ -0,0 +1,78 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>tcp-source</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>tcp-source</name>
<description>tcp source apps</description>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>tcp-supplier</artifactId>
<version>${java-functions.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>tcp</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.tcp.TcpSupplierConfiguration.class</configClass>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>tcp-supplier</artifactId>
<version>${java-functions.version}</version>
</dependency>
</dependencies>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</repositories>
</project>

2
applications/source/tcp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -0,0 +1,2 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.tcp.TcpSupplierProperties,\
org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties

98
applications/source/tcp-source/src/test/java/org/springframework/cloud/stream/app/source/tcp/TcpSourceTests.java Normal file
View File

@@ -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<byte[]> 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 {
}
}

37
functions/common/tcp-common/pom.xml Normal file
View File

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

108
functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java Normal file
View File

@@ -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<AbstractByteArraySerializer>
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;
}
}

56
functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java Normal file
View File

@@ -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;
}

96
functions/common/tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java Normal file
View File

@@ -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;
}
}

2
functions/pom.xml
View File

@@ -42,6 +42,7 @@
<modules>
<module>common/ftp-common</module>
<module>common/function-test-support</module>
<module>common/tcp-common</module>
<module>consumer/cassandra-consumer</module>
<module>consumer/counter-consumer</module>
@@ -64,6 +65,7 @@
<module>supplier/http-supplier</module>
<module>supplier/jdbc-supplier</module>
<module>supplier/mongodb-supplier</module>
<module>supplier/tcp-supplier</module>
<module>supplier/time-supplier</module>
<module>spring-functions-parent</module>

52
functions/supplier/tcp-supplier/pom.xml Normal file
View File

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

98
functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java Normal file
View File

@@ -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<Flux<Message<?>>> 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;
}
}

63
functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java Normal file
View File

@@ -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;
}
}

111
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java Normal file
View File

@@ -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<Flux<Message<?>>> tcpSupplier;
@Autowired
protected AbstractServerConnectionFactory connectionFactory;
@Autowired
protected TcpSupplierProperties properties;
/*
* Sends two messages with <prefix><payload><suffix> and asserts the
* payload is received on the other side.
*/
protected void doTest(String prefix, String payload, String suffix) throws Exception {
final Flux<Message<?>> 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 {
}
}

30
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/CRLFTests.java Normal file
View File

@@ -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");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L1Tests.java Normal file
View File

@@ -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", "");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L2Tests.java Normal file
View File

@@ -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", "");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/L4Tests.java Normal file
View File

@@ -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", "");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/LFTests.java Normal file
View File

@@ -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");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NULLTests.java Normal file
View File

@@ -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");
}
}

38
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/NotNioTests.java Normal file
View File

@@ -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);
}
}

43
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/PropertiesPopulatedTests.java Normal file
View File

@@ -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);
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/RAWTests.java Normal file
View File

@@ -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", "");
}
}

33
functions/supplier/tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/STXETXTests.java Normal file
View File

@@ -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");
}
}