Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

More LogAccessor; some new Sonar smells

Browse Source
This commit is contained in:
Artem Bilan
2020-10-06 16:31:43 -04:00
parent fe0e54c46d
commit bc63a0dac8
13 changed files with 105 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java
View File

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

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

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

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -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()) {

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java
View File

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

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java
View File

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

75
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java
View File

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

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java
View File

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

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java
View File

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

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java
View File

@@ -28,6 +28,8 @@ import org.springframework.integration.mapping.MessageMappingException;
* Writes a byte[] to an OutputStream prefixed by &lt;stx&gt; terminated by &lt;etx&gt;
*
* @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;
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -411,7 +411,7 @@ public class UnicastSendingMessageHandler extends
}
@Override
public void setLocalAddress(String localAddress) {
public synchronized void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}

2
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

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

62
spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java
View File

@@ -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 <foo@bar>\r\n"
+ "cc: a@b, c@d\r\n"
+ "bcc: e@f, g@h\r\n"
+ "From: Bar <bar@baz>\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 <bar@baz>\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
}
}
}
}