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`**
This commit is contained in:
shvo123
2022-01-02 17:56:22 -05:00
committed by Artem Bilan
parent 6c769cab37
commit 0b38b8df60
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-2021 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.
@@ -56,6 +56,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author John Anderson
* @author Artem Bilan
* @author David Herschler Shvo
*
* @since 2.0
*
@@ -74,7 +75,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();
@@ -115,7 +116,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) {
@@ -126,20 +126,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
@@ -376,9 +381,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);
@@ -394,7 +398,6 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
return null;
}
return message;
}
private void sendToChannel(Message<?> message) {
@@ -486,7 +489,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
catch (RejectedExecutionException e) {
this.executionControl.decrementAndGet();
logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " +
"this task executor pool size");
"this task executor pool size");
throw e;
}
}
@@ -619,12 +622,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