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

Fix new Sonar smells

Browse Source
This commit is contained in:
Artem Bilan
2020-03-04 10:41:45 -05:00
parent 0efcf2402a
commit e71e250e36
7 changed files with 115 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-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.
@@ -22,11 +22,15 @@ import java.io.PushbackInputStream;
import java.net.Socket;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.lang.Nullable;
/**
* Default implementation of {@link TcpNetConnectionSupport}.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 5.0
*
*/
@@ -34,7 +38,8 @@ public class DefaultTcpNetConnectionSupport extends AbstractTcpConnectionSupport
@Override
public TcpNetConnection createNewConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
if (isPushbackCapable()) {
return new PushBackTcpNetConnection(socket, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());
@@ -55,7 +60,9 @@ public class DefaultTcpNetConnectionSupport extends AbstractTcpConnectionSupport
private volatile InputStream wrapped;
PushBackTcpNetConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, int bufferSize) {
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName,
int bufferSize) {
super(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.pushbackBufferSize = bufferSize;
this.connectionId = "pushback:" + super.getConnectionId();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-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.
@@ -38,7 +38,8 @@ public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport
@Override
public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
if (isPushbackCapable()) {
return new PushBackTcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());
@@ -60,7 +61,7 @@ public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport
private volatile InputStream wrapped;
PushBackTcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
@Nullable ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
int bufferSize) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-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.
@@ -27,6 +27,7 @@ import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLParameters;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -75,7 +76,7 @@ public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupp
*/
@Override
public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
SSLEngine sslEngine = this.sslContext.createSSLEngine();
postProcessSSLEngine(sslEngine);
@@ -119,7 +120,8 @@ public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupp
private volatile InputStream wrapped;
PushBackTcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, SSLEngine sslEngine,
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName,
SSLEngine sslEngine,
int bufferSize) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName, sslEngine);

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2019 the original author or authors.
* Copyright 2001-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.
@@ -115,7 +115,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
* during event publishing, may be null, in which case "unknown" will be used.
*/
public TcpConnectionSupport(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
@Nullable ApplicationEventPublisher applicationEventPublisher,
@Nullable String connectionFactoryName) {
this.socketInfo = new SocketInfo(socket);
@@ -359,21 +359,15 @@ public abstract class TcpConnectionSupport implements TcpConnection {
}
protected void publishConnectionOpenEvent() {
TcpConnectionEvent event = new TcpConnectionOpenEvent(this,
getConnectionFactoryName());
doPublish(event);
doPublish(new TcpConnectionOpenEvent(this, getConnectionFactoryName()));
}
protected void publishConnectionCloseEvent() {
TcpConnectionEvent event = new TcpConnectionCloseEvent(this,
getConnectionFactoryName());
doPublish(event);
doPublish(new TcpConnectionCloseEvent(this, getConnectionFactoryName()));
}
protected void publishConnectionExceptionEvent(Throwable t) {
TcpConnectionEvent event = new TcpConnectionExceptionEvent(this,
getConnectionFactoryName(), t);
doPublish(event);
doPublish(new TcpConnectionExceptionEvent(this, getConnectionFactoryName(), t));
}
/**

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2019 the original author or authors.
* Copyright 2001-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.
@@ -42,6 +42,7 @@ import org.springframework.util.Assert;
* A TcpConnection that uses and underlying {@link Socket}.
*
* @author Gary Russell
*
* @since 2.0
*
*/
@@ -68,7 +69,8 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
* during event publishing, may be null, in which case "unknown" will be used.
*/
public TcpNetConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
@Nullable ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) {
super(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.socket = socket;
}

32
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-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.
@@ -53,6 +53,8 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author John Anderson
* @author Artem Bilan
*
* @since 2.0
*
*/
@@ -106,7 +108,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
* @param connectionFactoryName The name of the connection factory creating this connection.
*/
public TcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
@Nullable ApplicationEventPublisher applicationEventPublisher,
@Nullable String connectionFactoryName) {
super(socketChannel.socket(), server, lookupHost, applicationEventPublisher, connectionFactoryName);
@@ -272,8 +274,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
getConnectionId(), e);
logger.error("Read exception " + getConnectionId(), e);
}
else if (!isNoReadErrorOnClose()) {
logger.error("Read exception " +
@@ -386,13 +387,11 @@ public class TcpNioConnection extends TcpConnectionSupport {
private void sendToChannel(Message<?> message) {
try {
if (message != null) {
TcpListener listener = getListener();
if (listener == null) {
throw new NoListenerException("No listener");
}
listener.onMessage(message);
TcpListener listener = getListener();
if (listener == null) {
throw new NoListenerException("No listener");
}
listener.onMessage(message);
}
catch (Exception e) {
if (e instanceof NoListenerException) { // could also be thrown by an interceptor
@@ -425,7 +424,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
checkForAssembler();
if (logger.isTraceEnabled()) {
logger.trace("Before read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
logger.trace("Before read: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
int len = this.socketChannel.read(this.rawBuffer);
if (len < 0) {
@@ -433,11 +432,11 @@ public class TcpNioConnection extends TcpConnectionSupport {
closeConnection(true);
}
if (logger.isTraceEnabled()) {
logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
logger.trace("After read: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
this.rawBuffer.flip();
if (logger.isTraceEnabled()) {
logger.trace("After flip:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
logger.trace("After flip: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
if (logger.isDebugEnabled()) {
logger.debug("Read " + this.rawBuffer.limit() + " into raw buffer");
@@ -506,9 +505,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
closeConnection(true);
}
catch (Exception e) {
logger.error("Exception on Read " +
getConnectionId() + " " +
e.getMessage(), e);
logger.error("Exception on Read " + getConnectionId() + " " + e.getMessage(), e);
closeConnection(true);
}
}
@@ -535,8 +532,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
/**
* If true, connection will attempt to use direct buffers where
* possible.
* If true, connection will attempt to use direct buffers where possible.
* @param usingDirectBuffers the usingDirectBuffers to set.
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {

151
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-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.
@@ -17,7 +17,6 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Semaphore;
@@ -27,7 +26,6 @@ import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLEngineResult.Status;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLHandshakeException;
import javax.net.ssl.SSLSession;
@@ -63,26 +61,26 @@ public class TcpNioSSLConnection extends TcpNioConnection {
private final SSLEngine sslEngine;
private volatile ByteBuffer decoded;
private ByteBuffer decoded;
private volatile ByteBuffer encoded;
private volatile SSLChannelOutputStream sslChannelOutputStream;
private ByteBuffer encoded;
private final Semaphore semaphore = new Semaphore(0);
private final Object monitorLock = new Object();
private volatile boolean writerActive;
private volatile int handshakeTimeout = DEFAULT_HANDSHAKE_TIMEOUT;
private int handshakeTimeout = DEFAULT_HANDSHAKE_TIMEOUT;
private boolean needMoreNetworkData;
private SSLHandshakeException sslFatal;
private volatile SSLChannelOutputStream sslChannelOutputStream;
private volatile boolean writerActive;
public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
@Nullable ApplicationEventPublisher applicationEventPublisher, @Nullable String connectionFactoryName,
SSLEngine sslEngine) {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);
@@ -115,7 +113,8 @@ public class TcpNioSSLConnection extends TcpNioConnection {
protected void sendToPipe(final ByteBuffer networkBuffer) throws IOException {
Assert.notNull(networkBuffer, "rawBuffer cannot be null");
if (logger.isDebugEnabled()) {
logger.debug("sendToPipe " + this.sslEngine.getHandshakeStatus() + ", remaining: " + networkBuffer.remaining());
logger.debug("sendToPipe " + this.sslEngine.getHandshakeStatus() + ", remaining: " +
networkBuffer.remaining());
}
SSLEngineResult result = null;
while (!this.needMoreNetworkData) {
@@ -132,7 +131,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
}
}
this.needMoreNetworkData = false;
if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
if (Status.BUFFER_UNDERFLOW == result.getStatus()) {
networkBuffer.compact();
}
else {
@@ -155,53 +154,55 @@ public class TcpNioSSLConnection extends TcpNioConnection {
SSLEngineResult result = new SSLEngineResult(Status.OK, this.sslEngine.getHandshakeStatus(), 0, 0);
HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus();
switch (handshakeStatus) {
case NEED_TASK:
runTasks();
break;
case NEED_UNWRAP:
case FINISHED:
case NOT_HANDSHAKING:
this.decoded.clear();
result = this.sslEngine.unwrap(networkBuffer, this.decoded);
if (logger.isDebugEnabled()) {
logger.debug("After unwrap: " + resultToString(result));
}
Status status = result.getStatus();
if (status == Status.BUFFER_OVERFLOW) {
this.decoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize());
}
if (result.bytesProduced() > 0) {
this.decoded.flip();
super.sendToPipe(this.decoded);
}
break;
case NEED_WRAP:
if (!resumeWriterIfNeeded()) {
this.encoded.clear();
result = this.sslEngine.wrap(networkBuffer, this.encoded);
case NEED_TASK:
runTasks();
break;
case NEED_UNWRAP:
case FINISHED:
case NOT_HANDSHAKING:
this.decoded.clear();
result = this.sslEngine.unwrap(networkBuffer, this.decoded);
if (logger.isDebugEnabled()) {
logger.debug("After wrap: " + resultToString(result));
logger.debug("After unwrap: " + resultToString(result));
}
if (result.getStatus() == Status.BUFFER_OVERFLOW) {
this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize());
Status status = result.getStatus();
if (status == Status.BUFFER_OVERFLOW) {
this.decoded =
this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize());
}
else {
this.encoded.flip();
getSSLChannelOutputStream().writeEncoded(this.encoded);
if (result.bytesProduced() > 0) {
this.decoded.flip();
super.sendToPipe(this.decoded);
}
}
break;
default:
break;
case NEED_WRAP:
if (!resumeWriterIfNeeded()) {
this.encoded.clear();
result = this.sslEngine.wrap(networkBuffer, this.encoded);
if (logger.isDebugEnabled()) {
logger.debug("After wrap: " + resultToString(result));
}
if (result.getStatus() == Status.BUFFER_OVERFLOW) {
this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize());
}
else {
this.encoded.flip();
getSSLChannelOutputStream().writeEncoded(this.encoded);
}
}
break;
default:
}
switch (result.getHandshakeStatus()) {
case FINISHED:
resumeWriterIfNeeded(); //NOSONAR - fall-through intended
// switch fall-through intended
case NOT_HANDSHAKING:
case NEED_UNWRAP:
this.needMoreNetworkData = result.getStatus() == Status.BUFFER_UNDERFLOW || networkBuffer.remaining() == 0;
break;
default:
case FINISHED:
resumeWriterIfNeeded(); //NOSONAR - fall-through intended
// switch fall-through intended
case NOT_HANDSHAKING:
case NEED_UNWRAP:
this.needMoreNetworkData = result.getStatus() == Status.BUFFER_UNDERFLOW || networkBuffer
.remaining() == 0;
break;
default:
}
return result;
}
@@ -235,14 +236,12 @@ public class TcpNioSSLConnection extends TcpNioConnection {
/**
* Determines whether {@link #runTasks()} is needed and invokes if so.
*/
private HandshakeStatus runTasksIfNeeded(SSLEngineResult result) throws IOException {
if (result != null) {
if (logger.isDebugEnabled()) {
logger.debug("Running tasks if needed " + resultToString(result));
}
if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
runTasks();
}
private HandshakeStatus runTasksIfNeeded(SSLEngineResult result) {
if (logger.isDebugEnabled()) {
logger.debug("Running tasks if needed " + resultToString(result));
}
if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) {
runTasks();
}
HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus();
if (logger.isDebugEnabled()) {
@@ -258,12 +257,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
if (this.decoded == null) {
this.decoded = allocateEncryptionBuffer(2048);
this.encoded = allocateEncryptionBuffer(2048);
try {
initilizeEngine();
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
initializeEngine();
}
}
@@ -276,8 +270,8 @@ public class TcpNioSSLConnection extends TcpNioConnection {
}
}
private void initilizeEngine() throws IOException {
boolean client = !this.isServer();
private void initializeEngine() {
boolean client = !isServer();
this.sslEngine.setUseClientMode(client);
}
@@ -293,7 +287,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
protected SSLChannelOutputStream getSSLChannelOutputStream() {
if (this.sslChannelOutputStream == null) {
return (SSLChannelOutputStream) this.getChannelOutputStream();
return (SSLChannelOutputStream) getChannelOutputStream();
}
else {
return this.sslChannelOutputStream;
@@ -316,11 +310,11 @@ public class TcpNioSSLConnection extends TcpNioConnection {
* send to encrypted data to the SocketChannel.
*
*/
final class SSLChannelOutputStream extends ChannelOutputStream {
final class SSLChannelOutputStream extends ChannelOutputStream {
private final ChannelOutputStream channelOutputStream;
private SSLChannelOutputStream(ChannelOutputStream channelOutputStream) {
SSLChannelOutputStream(ChannelOutputStream channelOutputStream) {
this.channelOutputStream = channelOutputStream;
}
@@ -381,7 +375,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
result = encode(plainText);
status = result.getHandshakeStatus();
if (status == HandshakeStatus.NOT_HANDSHAKING ||
status == HandshakeStatus.FINISHED) {
status == HandshakeStatus.FINISHED) {
break;
}
}
@@ -428,17 +422,18 @@ public class TcpNioSSLConnection extends TcpNioConnection {
/**
* Encrypts plain text data. The result may indicate handshaking is needed.
*/
private SSLEngineResult encode(ByteBuffer plainText)
throws SSLException, IOException {
private SSLEngineResult encode(ByteBuffer plainText) throws IOException {
TcpNioSSLConnection.this.encoded.clear();
SSLEngineResult result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded);
SSLEngineResult result =
TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded);
if (logger.isDebugEnabled()) {
logger.debug("After wrap: "
+ resultToString(result)
+ " Plaintext buffer @" + plainText.position() + "/" + plainText.limit());
}
if (result.getStatus() == SSLEngineResult.Status.BUFFER_OVERFLOW) {
TcpNioSSLConnection.this.encoded = allocateEncryptionBuffer(TcpNioSSLConnection.this.sslEngine.getSession().getPacketBufferSize());
TcpNioSSLConnection.this.encoded =
allocateEncryptionBuffer(TcpNioSSLConnection.this.sslEngine.getSession().getPacketBufferSize());
result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded);
}
return result;