diff --git a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java index 3893c388..650698df 100644 --- a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java +++ b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/EncoderDecoderFactoryBean.java @@ -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 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); diff --git a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java index da6ca116..54ad1577 100644 --- a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java +++ b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/Encoding.java @@ -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 */ diff --git a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java index 54f17298..1ec3105a 100644 --- a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java +++ b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/TcpConnectionFactoryProperties.java @@ -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; diff --git a/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/package-info.java b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/package-info.java new file mode 100644 index 00000000..98d08fcf --- /dev/null +++ b/common/spring-tcp-common/src/main/java/org/springframework/cloud/fn/common/tcp/package-info.java @@ -0,0 +1,4 @@ +/** + * The TCP protocol supporting classes. + */ +package org.springframework.cloud.fn.common.tcp; diff --git a/consumer/spring-mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java b/consumer/spring-mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java index 320f3a59..6f33300f 100644 --- a/consumer/spring-mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java +++ b/consumer/spring-mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java @@ -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); } diff --git a/consumer/spring-tcp-consumer/README.adoc b/consumer/spring-tcp-consumer/README.adoc index 4d8ff29e..0f17aa58 100644 --- a/consumer/spring-tcp-consumer/README.adoc +++ b/consumer/spring-tcp-consumer/README.adoc @@ -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> 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 diff --git a/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java index 88b72117..7b1933e7 100644 --- a/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java +++ b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerConfiguration.java @@ -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()); diff --git a/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java index c17bdb8f..96bc7d53 100644 --- a/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java +++ b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/TcpConsumerProperties.java @@ -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) { diff --git a/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/package-info.java b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/package-info.java new file mode 100644 index 00000000..fdc988c4 --- /dev/null +++ b/consumer/spring-tcp-consumer/src/main/java/org/springframework/cloud/fn/consumer/tcp/package-info.java @@ -0,0 +1,4 @@ +/** + * The TCP consumer auto-configuration support. + */ +package org.springframework.cloud.fn.consumer.tcp; diff --git a/consumer/spring-tcp-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/consumer/spring-tcp-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..baeddf7b --- /dev/null +++ b/consumer/spring-tcp-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +org.springframework.cloud.fn.consumer.tcp.TcpConsumerConfiguration diff --git a/consumer/spring-tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java b/consumer/spring-tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java index 07b8392e..31b48805 100644 --- a/consumer/spring-tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java +++ b/consumer/spring-tcp-consumer/src/test/java/org/springframework/cloud/fn/consumer/tcp/AbstractTcpConsumerTests.java @@ -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) { } } diff --git a/supplier/spring-mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java b/supplier/spring-mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java index e559276c..8b19ef39 100644 --- a/supplier/spring-mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java +++ b/supplier/spring-mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java @@ -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); } diff --git a/supplier/spring-tcp-supplier/README.adoc b/supplier/spring-tcp-supplier/README.adoc index a957d350..2c730f9b 100644 --- a/supplier/spring-tcp-supplier/README.adoc +++ b/supplier/spring-tcp-supplier/README.adoc @@ -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>>`. diff --git a/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java index b3f4a256..af01d9a4 100644 --- a/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java +++ b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java @@ -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> 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>> tcpSupplier(Publisher> tcpSupplierFlow, - TcpReceivingChannelAdapter tcpReceivingChannelAdapter) { - - return () -> Flux.from(tcpSupplierFlow).doOnSubscribe(subscription -> tcpReceivingChannelAdapter.start()); + public Supplier>> tcpSupplier(Publisher> tcpSupplierFlow) { + return () -> Flux.from(tcpSupplierFlow); } } diff --git a/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java index d1dafe40..2406a0bc 100644 --- a/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java +++ b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierProperties.java @@ -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) { diff --git a/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/package-info.java b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/package-info.java new file mode 100644 index 00000000..004a4cea --- /dev/null +++ b/supplier/spring-tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/package-info.java @@ -0,0 +1,4 @@ +/** + * The TCP supplier auto-configuration support. + */ +package org.springframework.cloud.fn.supplier.tcp; diff --git a/supplier/spring-tcp-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/supplier/spring-tcp-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..2150d1de --- /dev/null +++ b/supplier/spring-tcp-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +org.springframework.cloud.fn.supplier.tcp.TcpSupplierConfiguration diff --git a/supplier/spring-tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java b/supplier/spring-tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java index 627b5425..c8e5270a 100644 --- a/supplier/spring-tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java +++ b/supplier/spring-tcp-supplier/src/test/java/org/springframework/cloud/fn/supplier/tcp/AbstractTcpSupplierTests.java @@ -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> 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();