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

GH-3705: Close TcpNioConn.ChannelOutStr.selector

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3705

Closing/destroying `ChannelOutputStream` object does not close the selector therefore it retains redundant pipes/FD that cen be seen using lsof command or ls /proc/

* Close `TcpNioConnection.ChannelOutputStream.selector` in the `ChannelOutputStream`
* Close `TcpNioConnection.ChannelOutputStream` when connection is closed
* Code style clean up

**Cherry-pick to `5.3.x` & `5.4.x`**

# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
This commit is contained in:
shvo123
2022-01-02 17:56:22 -05:00
committed by Artem Bilan
parent f6c6a7217b
commit d1be409cc7
1 changed files with 17 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2022 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.
@@ -54,6 +54,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author John Anderson
* @author Artem Bilan
* @author David Herschler Shvo
*
* @since 2.0
*
@@ -70,7 +71,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
private final SocketChannel socketChannel;
private final ChannelOutputStream channelOutputStream;
private final ChannelOutputStream channelOutputStream = new ChannelOutputStream();
private final ChannelInputStream channelInputStream = new ChannelInputStream();
@@ -113,7 +114,6 @@ public class TcpNioConnection extends TcpConnectionSupport {
super(socketChannel.socket(), server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.socketChannel = socketChannel;
this.channelOutputStream = new ChannelOutputStream();
}
public void setPipeTimeout(long pipeTimeout) {
@@ -124,20 +124,25 @@ public class TcpNioConnection extends TcpConnectionSupport {
public void close() {
setNoReadErrorOnClose(true);
doClose();
super.close();
}
private void doClose() {
try {
this.channelInputStream.close();
}
catch (@SuppressWarnings(UNUSED) IOException e) {
catch (@SuppressWarnings(UNUSED) IOException ex) {
}
try {
this.channelOutputStream.close();
}
catch (@SuppressWarnings(UNUSED) Exception ex) {
}
try {
this.socketChannel.close();
}
catch (@SuppressWarnings(UNUSED) Exception e) {
catch (@SuppressWarnings(UNUSED) Exception ex) {
}
super.close();
}
@Override
@@ -364,9 +369,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
throw new IOException("Interrupted waiting for IO", e);
}
}
Message<?> message = null;
try {
message = getMapper().toMessage(this);
return getMapper().toMessage(this);
}
catch (Exception e) {
closeConnection(true);
@@ -382,7 +386,6 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
return null;
}
return message;
}
private void sendToChannel(Message<?> message) {
@@ -477,7 +480,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
this.executionControl.decrementAndGet();
if (logger.isInfoEnabled()) {
logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " +
"this task executor pool size");
"this task executor pool size");
}
throw e;
}
@@ -609,12 +612,10 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
@Override
public void close() {
doClose();
}
@Override
public void flush() {
public void close() throws IOException {
if (this.selector != null) {
this.selector.close();
}
}
@Override