diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java index 4f42a72d14..267f15fd4c 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java @@ -43,6 +43,11 @@ import org.springframework.util.Assert; public abstract class FileTailingMessageProducerSupport extends MessageProducerSupport implements ApplicationEventPublisherAware { + /** + * The default delay between tail attempts in milliseconds. + */ + public static final long DEFAULT_TAIL_ATTEMPTS_DELAY = 5000L; + private final AtomicLong lastNoMessageAlert = new AtomicLong(); private File file; @@ -51,7 +56,7 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - private long tailAttemptsDelay = 5000; + private long tailAttemptsDelay = DEFAULT_TAIL_ATTEMPTS_DELAY; private long idleEventInterval = 0; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index cee6380abb..0d38e65724 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -41,6 +41,7 @@ import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.integration.ip.tcp.connection.TcpNioConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpSender; import org.springframework.integration.support.management.ManageableLifecycle; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; @@ -235,10 +236,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler } catch (RuntimeException | IOException ex) { logger.error(ex, "Tcp Gateway exception"); - if (ex instanceof MessagingException) { - throw (MessagingException) ex; - } - throw new MessagingException("Failed to send or receive", ex); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(requestMessage, + () -> "Failed to send or receive", ex); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index b45bef859a..187c8ea628 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -175,9 +175,9 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto } } } - catch (IOException ex) { // NOSONAR flow control via exceptions + catch (IOException ex) { // don't log an error if we had a good socket once and now it's closed - if (ex instanceof SocketException && theServerSocket != null) { + if (ex instanceof SocketException && theServerSocket != null) { // NOSONAR flow control via exceptions logger.info("Server Socket closed"); } else if (isActive()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java index 6a222d13d3..6a36e2347c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -18,11 +18,9 @@ package org.springframework.integration.ip.tcp.serializer; import java.io.IOException; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.core.log.LogAccessor; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -47,7 +45,7 @@ public abstract class AbstractByteArraySerializer implements */ public static final int DEFAULT_MAX_MESSAGE_SIZE = 2048; - protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR + protected final LogAccessor logger = new LogAccessor(this.getClass()); // NOSONAR private int maxMessageSize = DEFAULT_MAX_MESSAGE_SIZE; @@ -88,8 +86,8 @@ public abstract class AbstractByteArraySerializer implements if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent(event); } - else if (this.logger.isTraceEnabled()) { - this.logger.trace("No event publisher for " + event); + else { + this.logger.trace(() -> "No event publisher for " + event); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java index bcf89c562a..f568a9d6e5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -54,9 +54,8 @@ public class ByteArrayCrLfSerializer extends AbstractPooledBufferByteArraySerial public int fillToCrLf(InputStream inputStream, byte[] buffer) throws IOException { int n = 0; int bite; - if (logger.isDebugEnabled()) { - logger.debug("Available to read: " + inputStream.available()); - } + int available = inputStream.available(); + logger.debug(() -> "Available to read: " + available); try { while (true) { bite = inputStream.read(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java index 8ee59ef598..6521b048f4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java @@ -38,6 +38,8 @@ import java.nio.ByteBuffer; * {@link #writeHeader(OutputStream, int)}. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer { @@ -66,14 +68,14 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer private int headerAdjust; /** - * Constructs the serializer using {@link #HEADER_SIZE_INT} + * Construct the serializer using {@link #HEADER_SIZE_INT} */ public ByteArrayLengthHeaderSerializer() { this(HEADER_SIZE_INT); } /** - * Constructs the serializer using the supplied header size. + * Construct the serializer using the supplied header size. * Valid header sizes are {@link #HEADER_SIZE_INT} (default), * {@link #HEADER_SIZE_UNSIGNED_BYTE} and {@link #HEADER_SIZE_UNSIGNED_SHORT} * @param headerSize The header size. @@ -82,13 +84,13 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer if (headerSize != HEADER_SIZE_INT && headerSize != HEADER_SIZE_UNSIGNED_BYTE && headerSize != HEADER_SIZE_UNSIGNED_SHORT) { - throw new IllegalArgumentException("Illegal header size:" + headerSize); + throw new IllegalArgumentException("Illegal header size: " + headerSize); } this.headerSize = headerSize; } /** - * Return true if the lenght header value includes its own length. + * Return true if the length header value includes its own length. * @return true if the length includes the header length. * @since 5.2 */ @@ -124,45 +126,38 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer } /** - * Reads the header from the stream and then reads the provided length + * Read the header from the stream and then reads the provided length * from the stream and returns the data in a byte[]. Throws an * IOException if the length field exceeds the maxMessageSize. * Throws a {@link SoftEndOfStreamException} if the stream * is closed between messages. - * * @param inputStream The input stream. * @throws IOException Any IOException. */ @Override public byte[] deserialize(InputStream inputStream) throws IOException { int messageLength = this.readHeader(inputStream) - this.headerAdjust; - if (this.logger.isDebugEnabled()) { - this.logger.debug("Message length is " + messageLength); - } + this.logger.debug(() -> "Message length is " + messageLength); byte[] messagePart = null; try { - if (messageLength > getMaxMessageSize()) { + int maxMessageSize = getMaxMessageSize(); + if (messageLength > maxMessageSize) { throw new IOException("Message length " + messageLength + - " exceeds max message length: " + getMaxMessageSize()); + " exceeds max message length: " + maxMessageSize); } messagePart = new byte[messageLength]; read(inputStream, messagePart, false); return messagePart; } - catch (IOException e) { - publishEvent(e, messagePart, -1); - throw e; - } - catch (RuntimeException e) { - publishEvent(e, messagePart, -1); - throw e; + catch (IOException | RuntimeException ex) { + publishEvent(ex, messagePart, -1); + throw ex; } } /** - * Writes the byte[] to the output stream, preceded by a 4 byte + * Write the byte[] to the output stream, preceded by a 4 byte * length in network byte order (big endian). - * * @param bytes The bytes. * @param outputStream The output stream. */ @@ -173,9 +168,8 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer } /** - * Reads data from the socket and puts the data in buffer. Blocks until + * Read data from the socket and puts the data in buffer. Blocks until * buffer is full or a socket timeout occurs. - * * @param inputStream The input stream. * @param buffer the buffer into which the data should be read * @param header true if we are reading the header @@ -197,17 +191,15 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer throw new IOException("Stream closed after " + lengthRead + " of " + needed); } lengthRead += len; - if (this.logger.isDebugEnabled()) { - this.logger.debug("Read " + len + " bytes, buffer is now at " + - lengthRead + " of " + - needed); - } + int lengthReadToLog = lengthRead; + this.logger.debug(() -> "Read " + len + " bytes, buffer is now at " + + lengthReadToLog + " of " + needed); } return 0; } /** - * Writes the header, according to the header format. + * Write the header, according to the header format. * @param outputStream The output stream. * @param length The length. * @throws IOException Any IOException. @@ -220,29 +212,28 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer break; case HEADER_SIZE_UNSIGNED_BYTE: if (length > MAX_UNSIGNED_BYTE) { - throw new IllegalArgumentException("Length header:" + throw new IllegalArgumentException("Length header: " + this.headerSize - + " too short to accommodate message length:" + length); + + " too short to accommodate message length: " + length); } lengthPart.put((byte) length); break; case HEADER_SIZE_UNSIGNED_SHORT: if (length > MAX_UNSIGNED_SHORT) { - throw new IllegalArgumentException("Length header:" + throw new IllegalArgumentException("Length header: " + this.headerSize - + " too short to accommodate message length:" + length); + + " too short to accommodate message length: " + length); } lengthPart.putShort((short) length); break; default: - throw new IllegalArgumentException("Bad header size:" + this.headerSize); + throw new IllegalArgumentException("Bad header size: " + this.headerSize); } outputStream.write(lengthPart.array()); } /** - * Reads the header and returns the length of the data part. - * + * Read the header and returns the length of the data part. * @param inputStream The input stream. * @return The length of the data part. * @throws IOException Any IOException. @@ -261,7 +252,7 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer case HEADER_SIZE_INT: messageLength = ByteBuffer.wrap(lengthPart).getInt(); if (messageLength < 0) { - throw new IllegalArgumentException("Length header:" + throw new IllegalArgumentException("Length header: " + messageLength + " is negative"); } @@ -273,20 +264,16 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer messageLength = ByteBuffer.wrap(lengthPart).getShort() & MAX_UNSIGNED_SHORT; break; default: - throw new IllegalArgumentException("Bad header size:" + this.headerSize); + throw new IllegalArgumentException("Bad header size: " + this.headerSize); } return messageLength; } catch (SoftEndOfStreamException e) { // NOSONAR catch and throw throw e; // it's an IO exception and we don't want an event for this } - catch (IOException e) { - publishEvent(e, lengthPart, -1); - throw e; - } - catch (RuntimeException e) { - publishEvent(e, lengthPart, -1); - throw e; + catch (IOException | RuntimeException ex) { + publishEvent(ex, lengthPart, -1); + throw ex; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java index a6542cf4ff..ddb9c1cb85 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java @@ -75,9 +75,8 @@ public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySeriali protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException { int n = 0; int bite; - if (logger.isDebugEnabled()) { - logger.debug("Available to read: " + inputStream.available()); - } + int available = inputStream.available(); + logger.debug(() -> "Available to read: " + available); try { while (true) { try { @@ -95,8 +94,9 @@ public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySeriali } break; } - if (n >= getMaxMessageSize()) { - throw new IOException("Socket was not closed before max message length: " + getMaxMessageSize()); + int maxMessageSize = getMaxMessageSize(); + if (n >= maxMessageSize) { + throw new IOException("Socket was not closed before max message length: " + maxMessageSize); } buffer[n++] = (byte) bite; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java index 29a571f85f..314afea43f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java @@ -26,6 +26,8 @@ import java.io.OutputStream; * Writes a byte[] to an OutputStream and adds the terminator. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.2 */ public class ByteArraySingleTerminatorSerializer extends AbstractPooledBufferByteArraySerializer { @@ -46,9 +48,8 @@ public class ByteArraySingleTerminatorSerializer extends AbstractPooledBufferByt protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException { int n = 0; int bite; - if (logger.isDebugEnabled()) { - logger.debug("Available to read:" + inputStream.available()); - } + int available = inputStream.available(); + logger.debug(() -> "Available to read: " + available); try { while (true) { bite = inputStream.read(); @@ -60,10 +61,11 @@ public class ByteArraySingleTerminatorSerializer extends AbstractPooledBufferByt break; } buffer[n++] = (byte) bite; - if (n >= getMaxMessageSize()) { + int maxMessageSize = getMaxMessageSize(); + if (n >= maxMessageSize) { throw new IOException("Terminator '0x" + Integer.toHexString(this.terminator & 0xff) + "' not found before max message length: " - + getMaxMessageSize()); + + maxMessageSize); } } return copyToSizedArray(buffer, n); @@ -71,13 +73,9 @@ public class ByteArraySingleTerminatorSerializer extends AbstractPooledBufferByt catch (SoftEndOfStreamException e) { // NOSONAR catch and throw throw e; // it's an IO exception and we don't want an event for this } - catch (IOException e) { - publishEvent(e, buffer, n); - throw e; - } - catch (RuntimeException e) { - publishEvent(e, buffer, n); - throw e; + catch (IOException | RuntimeException ex) { + publishEvent(ex, buffer, n); + throw ex; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java index b6f9a4f1ab..5ffd7d3cff 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java @@ -28,6 +28,8 @@ import org.springframework.integration.mapping.MessageMappingException; * Writes a byte[] to an OutputStream prefixed by <stx> terminated by <etx> * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class ByteArrayStxEtxSerializer extends AbstractPooledBufferByteArraySerializer { @@ -63,20 +65,16 @@ public class ByteArrayStxEtxSerializer extends AbstractPooledBufferByteArraySeri while ((bite = inputStream.read()) != ETX) { checkClosure(bite); buffer[n++] = (byte) bite; - if (n >= getMaxMessageSize()) { - throw new IOException("ETX not found before max message length: " - + getMaxMessageSize()); + int maxMessageSize = getMaxMessageSize(); + if (n >= maxMessageSize) { + throw new IOException("ETX not found before max message length: " + maxMessageSize); } } return copyToSizedArray(buffer, n); } - catch (IOException e) { - publishEvent(e, buffer, n); - throw e; - } - catch (RuntimeException e) { - publishEvent(e, buffer, n); - throw e; + catch (IOException | RuntimeException ex) { + publishEvent(ex, buffer, n); + throw ex; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 599fb15891..50582b4089 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -136,7 +136,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece stop(); } catch (Exception ex) { - if (ex instanceof MessagingException) { + if (ex instanceof MessagingException) { // NOSONAR flow control via exceptions throw (MessagingException) ex; } throw new MessagingException("failed to receive DatagramPacket", ex); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 29e225348f..dd4ca5765f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -411,7 +411,7 @@ public class UnicastSendingMessageHandler extends } @Override - public void setLocalAddress(String localAddress) { + public synchronized void setLocalAddress(String localAddress) { this.localAddress = localAddress; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index ab1c26fa33..b00e56f09b 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -1311,7 +1311,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } } catch (Exception ex) { - logger.warn(() -> "Failed to consume reply with correlationId " + correlationId); + logger.warn(ex, () -> "Failed to consume reply with correlationId " + correlationId); } } diff --git a/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java b/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java index efadf92110..789ccb544f 100644 --- a/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java +++ b/spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java @@ -183,31 +183,29 @@ public final class TestMailServer { write("+OK POP3"); while (!socket.isClosed()) { String line = reader.readLine(); - if ("CAPA".equals(line)) { - write(PLUS_OK); - write("USER"); - write("."); - } - else if ("USER user".equals(line)) { - write(PLUS_OK); - } - else if ("PASS pw".equals(line)) { - write(PLUS_OK); - } - else if ("STAT".equals(line)) { - write("+OK 1 3"); - } - else if ("NOOP".equals(line)) { - write(PLUS_OK); - } - else if ("RETR 1".equals(line)) { - write(PLUS_OK); - write(MESSAGE); - write("."); - } - else if ("QUIT".equals(line)) { - write(PLUS_OK); - socket.close(); + switch (line) { + case "CAPA": + write(PLUS_OK); + write("USER"); + write("."); + break; + case "USER user": + case "PASS pw": + case "NOOP": + write(PLUS_OK); + break; + case "STAT": + write("+OK 1 3"); + break; + case "RETR 1": + write(PLUS_OK); + write(MESSAGE); + write("."); + break; + case "QUIT": + write(PLUS_OK); + socket.close(); + break; } } } @@ -257,7 +255,8 @@ public final class TestMailServer { super(socket); } - @Override // NOSONAR + @Override + // NOSONAR void doRun() { try { write("* OK IMAP4rev1 Service Ready"); @@ -484,11 +483,11 @@ public final class TestMailServer { public static final String MESSAGE = "To: Foo \r\n" - + "cc: a@b, c@d\r\n" - + "bcc: e@f, g@h\r\n" - + "From: Bar \r\n" - + "Subject: Test Email\r\n" - + "\r\n" + BODY; + + "cc: a@b, c@d\r\n" + + "bcc: e@f, g@h\r\n" + + "From: Bar \r\n" + + "Subject: Test Email\r\n" + + "\r\n" + BODY; protected final Socket socket; // NOSONAR protected @@ -533,6 +532,7 @@ public final class TestMailServer { // NOSONAR } } + } }