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

TCP Sink

Browse Source 
Migrating TCP Sink as a consumer/sink.
This commit is contained in:
Soby Chacko
2020-05-18 16:00:36 -04:00
committed by Artem Bilan
parent dfeef686b3
commit 5da0cc8cfb
20 changed files with 1094 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/sink/pom.xml
View File

@@ -21,6 +21,7 @@
<module>rabbit-sink</module>
<module>router-sink</module>
<module>sftp-sink</module>
<module>tcp-sink</module>
<module>throughput-sink</module>
</modules>
</project>

41
applications/sink/tcp-sink/README.adoc Normal file
View File

@@ -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: `$$<none>$$`, possible values: `CRLF`,`LF`,`NULL`,`STXETX`,`RAW`,`L1`,`L2`,`L4`)*
$$tcp.consumer.host$$:: $$The host to which this sink will connect.$$ *($$String$$, default: `$$<none>$$`)*
$$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[]

78
applications/sink/tcp-sink/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-sink</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>tcp-sink</name>
<description>tcp sink 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-consumer</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>sink</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.consumer.tcp.TcpConsumerConfiguration.class</configClass>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>tcp-consumer</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>

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

@@ -0,0 +1,3 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.tcp.TcpConsumerProperties,\
org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties

181
applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java Normal file
View File

@@ -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<String> 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<String> 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<String> queue = new LinkedBlockingQueue<>();
private volatile boolean stopped;
TestTCPServer() {
ServerSocket serverSocket = null;
ExecutorService executor = null;
try {
serverSocket = ServerSocketFactory.getDefault().createServerSocket(0);
System.setProperty("tcp.consumer.test.port", Integer.toString(serverSocket.getLocalPort()));
executor = Executors.newSingleThreadExecutor();
}
catch (IOException e) {
e.printStackTrace();
}
this.serverSocket = serverSocket;
this.executor = executor;
this.decoder = new ByteArrayCrLfSerializer();
executor.execute(this);
}
private void setDecoder(AbstractByteArraySerializer decoder) {
this.decoder = decoder;
}
@Override
public void run() {
while (true) {
Socket socket = null;
try {
logger.info("Server listening on " + this.serverSocket.getLocalPort());
socket = this.serverSocket.accept();
while (true) {
byte[] data = decoder.deserialize(socket.getInputStream());
queue.offer(new String(data));
}
}
catch (SoftEndOfStreamException e) {
// normal close
}
catch (IOException e) {
try {
if (socket != null) {
socket.close();
}
}
catch (IOException e1) {
}
logger.error(e.getMessage());
if (this.stopped) {
logger.info("Server stopped on " + this.serverSocket.getLocalPort());
break;
}
}
}
}
private void shutDown() {
try {
this.stopped = true;
this.serverSocket.close();
this.executor.shutdownNow();
}
catch (IOException e) {
}
}
}
@SpringBootApplication
@Import(TcpConsumerConfiguration.class)
public static class TcpSinkTestApplication {
}
}

52
functions/consumer/tcp-consumer/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-consumer</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>tcp-consumer</name>
<description>tcp consumer</description>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>tcp-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

110
functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java Normal file
View File

@@ -0,0 +1,110 @@
/*
* Copyright 2015-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.consumer.tcp;
import java.util.function.Consumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.tcp.EncoderDecoderFactoryBean;
import org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.messaging.Message;
/**
* A consumer that sends data over TCP.
*
* @author Gary Russell
* @author Christian Tzolov
*/
@Configuration
@EnableConfigurationProperties({TcpConsumerProperties.class, TcpConnectionFactoryProperties.class})
public class TcpConsumerConfiguration {
@Autowired
private TcpConsumerProperties properties;
@Autowired
private TcpConnectionFactoryProperties tcpConnectionProperties;
@Qualifier("tcpSinkConnectionFactory")
@Autowired
private AbstractConnectionFactory connectionFactory;
@Bean
public Consumer<Message<?>> tcpConsumer() {
return handler()::handleMessage;
}
@Bean
public TcpSendingMessageHandlerSmartLifeCycle handler() {
TcpSendingMessageHandlerSmartLifeCycle tcpMessageHandler = new TcpSendingMessageHandlerSmartLifeCycle();
tcpMessageHandler.setConnectionFactory(connectionFactory);
return tcpMessageHandler;
}
@Bean
public TcpConnectionFactoryFactoryBean tcpSinkConnectionFactory(
@Qualifier("tcpSinkEncoder") AbstractByteArraySerializer encoder,
@Qualifier("tcpSinkMapper") TcpMessageMapper mapper) throws Exception {
TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
factoryBean.setType("client");
factoryBean.setHost(this.properties.getHost());
factoryBean.setPort(this.tcpConnectionProperties.getPort());
factoryBean.setUsingNio(this.tcpConnectionProperties.isNio());
factoryBean.setUsingDirectBuffers(this.tcpConnectionProperties.isUseDirectBuffers());
factoryBean.setLookupHost(this.tcpConnectionProperties.isReverseLookup());
factoryBean.setSerializer(encoder);
factoryBean.setSoTimeout(this.tcpConnectionProperties.getSocketTimeout());
factoryBean.setMapper(mapper);
factoryBean.setSingleUse(this.properties.isClose());
return factoryBean;
}
@Bean
public EncoderDecoderFactoryBean tcpSinkEncoder() {
return new EncoderDecoderFactoryBean(this.properties.getEncoder());
}
@Bean
public TcpMessageMapper tcpSinkMapper() {
TcpMessageMapper mapper = new TcpMessageMapper();
mapper.setCharset(this.properties.getCharset());
return mapper;
}
static class TcpSendingMessageHandlerSmartLifeCycle extends TcpSendingMessageHandler implements SmartLifecycle {
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public int getPhase() {
return Integer.MIN_VALUE;
}
}
}

90
functions/consumer/tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java Normal file
View File

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

182
functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java Normal file
View File

@@ -0,0 +1,182 @@
/*
* Copyright 2015-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.consumer.tcp;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for TCP Consumer.
*
* @author Gary Russell
* @author Soby Chacko
*/
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = { "tcp.consumer.host = localhost", "tcp.port = ${tcp.consumer.test.port}" })
@DirtiesContext
public class AbstractTcpConsumerTests {
private static TestTCPServer server;
@Autowired
protected AbstractClientConnectionFactory connectionFactory;
@Autowired
Consumer<Message<?>> tcpConsumer;
@BeforeAll
public static void startup() {
server = new TestTCPServer();
}
@AfterAll
public static void shutDown() {
server.shutDown();
}
/*
* Sends two messages and asserts they arrive as expected on the other side using
* the supplied decoder.
*/
protected void doTest(AbstractByteArraySerializer decoder) throws Exception {
server.setDecoder(decoder);
Message<String> message = new GenericMessage<>("foo");
tcpConsumer.accept(message);
String received = server.queue.poll(10, TimeUnit.SECONDS);
assertThat(received).isEqualTo("foo");
tcpConsumer.accept(message);
received = server.queue.poll(10, TimeUnit.SECONDS);
assertThat(received).isEqualTo("foo");
}
/**
* TCP server that uses the supplied {@link AbstractByteArraySerializer}
* to decode the input stream and put the resulting message in a queue.
*
*/
private static class TestTCPServer implements Runnable {
private static final Log logger = LogFactory.getLog(TestTCPServer.class);
private final ServerSocket serverSocket;
private final ExecutorService executor;
private volatile AbstractByteArraySerializer decoder;
private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
private volatile boolean stopped;
TestTCPServer() {
ServerSocket serverSocket = null;
ExecutorService executor = null;
try {
serverSocket = ServerSocketFactory.getDefault().createServerSocket(0);
System.setProperty("tcp.consumer.test.port", Integer.toString(serverSocket.getLocalPort()));
executor = Executors.newSingleThreadExecutor();
}
catch (IOException e) {
e.printStackTrace();
}
this.serverSocket = serverSocket;
this.executor = executor;
this.decoder = new ByteArrayCrLfSerializer();
executor.execute(this);
}
private void setDecoder(AbstractByteArraySerializer decoder) {
this.decoder = decoder;
}
@Override
public void run() {
while (true) {
Socket socket = null;
try {
logger.info("Server listening on " + this.serverSocket.getLocalPort());
socket = this.serverSocket.accept();
while (true) {
byte[] data = decoder.deserialize(socket.getInputStream());
queue.offer(new String(data));
}
}
catch (SoftEndOfStreamException e) {
// normal close
}
catch (IOException e) {
try {
if (socket != null) {
socket.close();
}
}
catch (IOException e1) {
}
logger.error(e.getMessage());
if (this.stopped) {
logger.info("Server stopped on " + this.serverSocket.getLocalPort());
break;
}
}
}
}
private void shutDown() {
try {
this.stopped = true;
this.serverSocket.close();
this.executor.shutdownNow();
}
catch (IOException e) {
}
}
}
@SpringBootApplication
public static class TcpConsumerTestApplication {
}
}

33
functions/consumer/tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/CRLFTests.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.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());
}
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1
functions/pom.xml
View File

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