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

checkstyle FinalClassCheck

Browse Source 
fixes

fixModifiers after fixFinal

Revert CachingSessionFactory

Class is spied in tests.

checkstyle - Import Rules

checkstyle InterfaceIsType

checkstyle InnerTypeLast

checkstyle OneStatementPerLine

CovariantEquals
OneTopLevelClass

* Revert `'\n'` -> `System.lineSeparator()` in the Gradle scripts to meet Git `autocrlf = true` on Windows
* Fix timing issue with the `LastModifiedFileListFilterTests`, when the `age = 1` might not be enough for the file object when we have some delay before checking
This commit is contained in:
Gary Russell
2016-04-02 10:52:19 -04:00
committed by Artem Bilan
parent 43af472c3a
commit 4ac3a79df7
105 changed files with 752 additions and 539 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -274,7 +274,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
* @author Gary Russell
* @since 2.0
*/
private class AsyncReply {
private final class AsyncReply {
private final CountDownLatch latch;

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -897,7 +897,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
+ ", port=" + getPort();
}
private class PendingIO {
private final class PendingIO {
private final long failedAt;

176
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -136,94 +136,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return new CachedConnection(this.pool.getItem(), getListener());
}
private class CachedConnection extends TcpConnectionInterceptorSupport {
private final AtomicBoolean released = new AtomicBoolean();
private CachedConnection(TcpConnectionSupport connection, TcpListener tcpListener) {
super.setTheConnection(connection);
registerListener(tcpListener);
}
@Override
public void close() {
if (!this.released.compareAndSet(false, true)) {
if (logger.isDebugEnabled()) {
logger.debug("Connection " + getConnectionId() + " has already been released");
}
}
else {
/**
* If the delegate is stopped, actually close the connection, but still release
* it to the pool, it will be discarded/renewed the next time it is retrieved.
*/
if (!isRunning()) {
if (logger.isDebugEnabled()) {
logger.debug("Factory not running - closing " + getConnectionId());
}
super.close();
}
CachingClientConnectionFactory.this.pool.releaseItem(getTheConnection());
}
}
@Override
public String getConnectionId() {
return "Cached:" + super.getConnectionId();
}
@Override
public String toString() {
return getConnectionId();
}
/**
* We have to intercept the message to replace the connectionId header with
* ours so the listener can correlate a response with a request. We supply
* the actual connectionId in another header for convenience and tracing
* purposes.
*/
@Override
public boolean onMessage(Message<?> message) {
Message<?> modifiedMessage;
if (message instanceof ErrorMessage) {
Map<String, Object> headers = new HashMap<String, Object>(message.getHeaders());
headers.put(IpHeaders.CONNECTION_ID, getConnectionId());
if (headers.get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
headers.put(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
modifiedMessage = new ErrorMessage((Throwable) message.getPayload(), headers);
}
else {
AbstractIntegrationMessageBuilder<?> messageBuilder =
CachingClientConnectionFactory.this.getMessageBuilderFactory()
.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
modifiedMessage = messageBuilder.build();
}
TcpListener listener = getListener();
if (listener != null) {
listener.onMessage(modifiedMessage);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Message discarded; no listener: " + message);
}
}
return true;
}
private void physicallyClose() {
getTheConnection().close();
}
}
///////////////// DELEGATE METHODS ///////////////////////
@Override
@@ -491,4 +403,92 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
this.targetConnectionFactory.stop(callback);
}
private final class CachedConnection extends TcpConnectionInterceptorSupport {
private final AtomicBoolean released = new AtomicBoolean();
private CachedConnection(TcpConnectionSupport connection, TcpListener tcpListener) {
super.setTheConnection(connection);
registerListener(tcpListener);
}
@Override
public void close() {
if (!this.released.compareAndSet(false, true)) {
if (logger.isDebugEnabled()) {
logger.debug("Connection " + getConnectionId() + " has already been released");
}
}
else {
/**
* If the delegate is stopped, actually close the connection, but still release
* it to the pool, it will be discarded/renewed the next time it is retrieved.
*/
if (!isRunning()) {
if (logger.isDebugEnabled()) {
logger.debug("Factory not running - closing " + getConnectionId());
}
super.close();
}
CachingClientConnectionFactory.this.pool.releaseItem(getTheConnection());
}
}
@Override
public String getConnectionId() {
return "Cached:" + super.getConnectionId();
}
@Override
public String toString() {
return getConnectionId();
}
/**
* We have to intercept the message to replace the connectionId header with
* ours so the listener can correlate a response with a request. We supply
* the actual connectionId in another header for convenience and tracing
* purposes.
*/
@Override
public boolean onMessage(Message<?> message) {
Message<?> modifiedMessage;
if (message instanceof ErrorMessage) {
Map<String, Object> headers = new HashMap<String, Object>(message.getHeaders());
headers.put(IpHeaders.CONNECTION_ID, getConnectionId());
if (headers.get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
headers.put(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
modifiedMessage = new ErrorMessage((Throwable) message.getPayload(), headers);
}
else {
AbstractIntegrationMessageBuilder<?> messageBuilder =
CachingClientConnectionFactory.this.getMessageBuilderFactory()
.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
modifiedMessage = messageBuilder.build();
}
TcpListener listener = getListener();
if (listener != null) {
listener.onMessage(modifiedMessage);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Message discarded; no listener: " + message);
}
}
return true;
}
private void physicallyClose() {
getTheConnection().close();
}
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -144,7 +144,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* @since 2.2
*
*/
private class FailoverTcpConnection extends TcpConnectionSupport implements TcpListener {
private final class FailoverTcpConnection extends TcpConnectionSupport implements TcpListener {
private final List<AbstractClientConnectionFactory> factories;

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

@@ -278,7 +278,7 @@ public class TcpNioSSLConnection extends TcpNioConnection {
* send to encrypted data to the SocketChannel.
*
*/
class SSLChannelOutputStream extends ChannelOutputStream {
final class SSLChannelOutputStream extends ChannelOutputStream {
private final ChannelOutputStream channelOutputStream;

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java
View File

@@ -28,7 +28,7 @@ import org.springframework.integration.ip.tcp.connection.AbstractServerConnectio
* @since 2.2
*
*/
public class TestingUtilities {
public final class TestingUtilities {
private TestingUtilities() {
super();