SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make TCP modules as auto-config

Browse Source 
* Fix their Checkstyle violations
This commit is contained in:
Artem Bilan
2024-01-05 15:11:55 -05:00
parent e1a1845dab
commit 76e7ed1b69
18 changed files with 81 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2024 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.
@@ -62,36 +62,17 @@ public class EncoderDecoderFactoryBean extends AbstractFactoryBean<AbstractByteA
}
@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);
}
protected AbstractByteArraySerializer createInstance() {
AbstractByteArraySerializer codec = switch (this.encoding) {
case CRLF -> new ByteArrayCrLfSerializer();
case LF -> new ByteArrayLfSerializer();
case NULL -> new ByteArraySingleTerminatorSerializer((byte) 0);
case STXETX -> new ByteArrayStxEtxSerializer();
case L1 -> new ByteArrayLengthHeaderSerializer(1);
case L2 -> new ByteArrayLengthHeaderSerializer(2);
case L4 -> new ByteArrayLengthHeaderSerializer(4);
case RAW -> new ByteArrayRawSerializer();
};
codec.setApplicationEventPublisher(this.applicationEventPublisher);
if (this.maxMessageSize != null) {
codec.setMaxMessageSize(this.maxMessageSize);

4
common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 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.
@@ -17,6 +17,8 @@
package org.springframework.cloud.fn.common.tcp;
/**
* The encoding modes.
*
* @author Gary Russell
* @author Christian Tzolov
*/

6
common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2024 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.
@@ -45,12 +45,12 @@ public class TcpConnectionFactoryProperties {
private int socketTimeout = 120000;
/**
* Whether or not to use NIO.
* Whether to use NIO.
*/
private boolean nio = false;
/**
* Whether or not to use direct buffers.
* Whether to use direct buffers.
*/
private boolean useDirectBuffers = false;

4
common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The TCP protocol supporting classes.
*/
package org.springframework.cloud.fn.common.tcp;

2
consumer/spring-mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java
View File

@@ -58,7 +58,7 @@ public class MqttConsumerTests implements MosquittoContainerTest {
protected QueueChannel queue;
@DynamicPropertySource
static void mongoDbProperties(DynamicPropertyRegistry registry) {
static void mqttConnectionProperties(DynamicPropertyRegistry registry) {
registry.add("mqtt.url", MosquittoContainerTest::mqttUrl);
}

4
consumer/spring-tcp-consumer/README.adoc
View File

@@ -4,7 +4,7 @@ A consumer that allows you to send TCP messages.
## Beans for injection
You can import `TcpConsumerConfiguration` in the application and then inject the following bean.
The `TcpConsumerConfiguration` auto-configuration provides the following bean:
`Consumer<Message<?>> tcpConsumer`
@@ -14,7 +14,7 @@ You can use `tcpConsumer` as a qualifier when injecting.
All configuration properties are prefixed with `tcp.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/tcp/TCPConsumerProperties.java[TcpConsumerProperties].
For more information on the various options available, please see `TCPConsumerProperties`.
## Tests

11
consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2024 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.
@@ -19,12 +19,12 @@ package org.springframework.cloud.fn.consumer.tcp;
import java.util.function.Consumer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
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;
@@ -39,7 +39,7 @@ import org.springframework.messaging.Message;
* @author Christian Tzolov
* @author Chris Bono
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration
@EnableConfigurationProperties({ TcpConsumerProperties.class, TcpConnectionFactoryProperties.class })
public class TcpConsumerConfiguration {
@@ -49,6 +49,7 @@ public class TcpConsumerConfiguration {
public TcpConsumerConfiguration(TcpConsumerProperties properties,
TcpConnectionFactoryProperties tcpConnectionProperties) {
this.properties = properties;
this.tcpConnectionProperties = tcpConnectionProperties;
}
@@ -61,6 +62,7 @@ public class TcpConsumerConfiguration {
@Bean
public TcpSendingMessageHandlerSmartLifeCycle handler(
@Qualifier("tcpSinkConnectionFactory") AbstractConnectionFactory connectionFactory) {
TcpSendingMessageHandlerSmartLifeCycle tcpMessageHandler = new TcpSendingMessageHandlerSmartLifeCycle();
tcpMessageHandler.setConnectionFactory(connectionFactory);
return tcpMessageHandler;
@@ -69,7 +71,8 @@ public class TcpConsumerConfiguration {
@Bean
public TcpConnectionFactoryFactoryBean tcpSinkConnectionFactory(
@Qualifier("tcpSinkEncoder") AbstractByteArraySerializer encoder,
@Qualifier("tcpSinkMapper") TcpMessageMapper mapper) throws Exception {
@Qualifier("tcpSinkMapper") TcpMessageMapper mapper) {
TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
factoryBean.setType("client");
factoryBean.setHost(this.properties.getHost());

8
consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2024 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.
@@ -55,7 +55,7 @@ public class TcpConsumerProperties {
@NotNull
public String getHost() {
return host;
return this.host;
}
public void setHost(String host) {
@@ -73,7 +73,7 @@ public class TcpConsumerProperties {
@NotNull
public String getCharset() {
return charset;
return this.charset;
}
public void setCharset(String charset) {
@@ -81,7 +81,7 @@ public class TcpConsumerProperties {
}
public boolean isClose() {
return close;
return this.close;
}
public void setClose(boolean close) {

4
consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The TCP consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.tcp;

1
consumer/spring-tcp-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.tcp.TcpConsumerConfiguration

25
consumer/spring-tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2024 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.
@@ -43,6 +43,8 @@ import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamExceptio
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@@ -52,8 +54,7 @@ import static org.assertj.core.api.Assertions.assertThat;
* @author Gary Russell
* @author Soby Chacko
*/
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = { "tcp.consumer.host = localhost", "tcp.port = ${tcp.consumer.test.port}" })
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = "tcp.consumer.host = localhost")
@DirtiesContext
public class AbstractTcpConsumerTests {
@@ -70,6 +71,11 @@ public class AbstractTcpConsumerTests {
server = new TestTCPServer();
}
@DynamicPropertySource
static void tcpConnectionProperties(DynamicPropertyRegistry registry) {
registry.add("tcp.port", () -> server.serverSocket.getLocalPort());
}
@AfterAll
public static void shutDown() {
server.shutDown();
@@ -115,11 +121,10 @@ public class AbstractTcpConsumerTests {
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();
catch (IOException ex) {
logger.error(ex);
}
this.serverSocket = serverSocket;
this.executor = executor;
@@ -143,10 +148,10 @@ public class AbstractTcpConsumerTests {
queue.offer(new String(data));
}
}
catch (SoftEndOfStreamException e) {
catch (SoftEndOfStreamException ex) {
// normal close
}
catch (IOException e) {
catch (IOException ex) {
try {
if (socket != null) {
socket.close();
@@ -154,7 +159,7 @@ public class AbstractTcpConsumerTests {
}
catch (IOException e1) {
}
logger.error(e.getMessage());
logger.error(ex.getMessage());
if (this.stopped) {
logger.info("Server stopped on " + this.serverSocket.getLocalPort());
break;
@@ -169,7 +174,7 @@ public class AbstractTcpConsumerTests {
this.serverSocket.close();
this.executor.shutdownNow();
}
catch (IOException e) {
catch (IOException ex) {
}
}

2
supplier/spring-mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java
View File

@@ -57,7 +57,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class MqttSupplierTests implements MosquittoContainerTest {
@DynamicPropertySource
static void mongoDbProperties(DynamicPropertyRegistry registry) {
static void mqttConnectionProperties(DynamicPropertyRegistry registry) {
registry.add("mqtt.url", MosquittoContainerTest::mqttUrl);
}

2
supplier/spring-tcp-supplier/README.adoc
View File

@@ -9,7 +9,7 @@ Users have to subscribe to this `Flux` and then receive the data.
## Beans for injection
You can import the `TcpSupplierConfiguration` in the application and then inject the following bean: `tcpSupplier`.
The `TcpSupplierConfiguration` auto-configuration provides the following bean: `tcpSupplier`.
You need to inject this as `Supplier<Flux<Message<?>>>`.

16
supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2023 the original author or authors.
* Copyright 2015-2024 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.
@@ -22,11 +22,11 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
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.dsl.IntegrationFlow;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
@@ -36,14 +36,14 @@ import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerial
import org.springframework.messaging.Message;
/**
* A source module that receives data over TCP.
* A supplier that receives data over TCP.
*
* @author Gary Russell
* @author Christian Tzolov
* @author Soby Chacko
* @author Artem Bilan
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration
@EnableConfigurationProperties({ TcpSupplierProperties.class, TcpConnectionFactoryProperties.class })
public class TcpSupplierConfiguration {
@@ -82,14 +82,12 @@ public class TcpSupplierConfiguration {
@Bean
public Publisher<Message<Object>> tcpSupplierFlow(TcpReceivingChannelAdapter adapter) {
return IntegrationFlow.from(adapter).headerFilter(IpHeaders.LOCAL_ADDRESS).toReactivePublisher();
return IntegrationFlow.from(adapter).headerFilter(IpHeaders.LOCAL_ADDRESS).toReactivePublisher(true);
}
@Bean
public Supplier<Flux<Message<Object>>> tcpSupplier(Publisher<Message<Object>> tcpSupplierFlow,
TcpReceivingChannelAdapter tcpReceivingChannelAdapter) {
return () -> Flux.from(tcpSupplierFlow).doOnSubscribe(subscription -> tcpReceivingChannelAdapter.start());
public Supplier<Flux<Message<Object>>> tcpSupplier(Publisher<Message<Object>> tcpSupplierFlow) {
return () -> Flux.from(tcpSupplierFlow);
}
}

6
supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2024 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.
@@ -23,7 +23,7 @@ import org.springframework.cloud.fn.common.tcp.Encoding;
import org.springframework.validation.annotation.Validated;
/**
* Properties for the TCP Source.
* Properties for the TCP supplier.
*
* @author Gary Russell
* @author Christian Tzolov
@@ -53,7 +53,7 @@ public class TcpSupplierProperties {
}
public int getBufferSize() {
return bufferSize;
return this.bufferSize;
}
public void setBufferSize(int bufferSize) {

4
supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The TCP supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.tcp;

1
supplier/spring-tcp-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.tcp.TcpSupplierConfiguration

16
supplier/spring-tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2021 the original author or authors.
* Copyright 2015-2024 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.
@@ -61,22 +61,22 @@ public class AbstractTcpSupplierTests {
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();
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) {
if (prefix.isEmpty() && suffix.isEmpty()) {
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) {
if (prefix.isEmpty() && suffix.isEmpty()) {
socket.close(); // RAW - for the others, close AFTER the messages are decoded.
}
socket.close();