GH-3705: Close TcpNioConn.ChannelOutStr.selector

Fixes #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

Author: shvo123

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

@@ -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 @@
  * @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 TcpNioConnection(SocketChannel socketChannel, boolean server, boolean loo
 
 		super(socketChannel.socket(), server, lookupHost, applicationEventPublisher, connectionFactoryName);
 		this.socketChannel = socketChannel;
-		this.channelOutputStream = new ChannelOutputStream();
 	}
 
 	public void setPipeTimeout(long pipeTimeout) {
@@ -124,20 +124,25 @@ public void setPipeTimeout(long pipeTimeout) {
 	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 @@ private synchronized Message<?> convert() throws IOException {
 				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 @@ private synchronized Message<?> convert() throws IOException {
 			}
 			return null;
 		}
-		return message;
 	}
 
 	private void sendToChannel(Message<?> message) {
@@ -477,7 +480,7 @@ private void checkForAssembler() {
 					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 void write(int b) throws IOException {
 		}
 
 		@Override
-		public void close() {
-			doClose();
-		}
-
-		@Override
-		public void flush() {
+		public void close() throws IOException {
+			if (this.selector != null) {
+				this.selector.close();
+			}
 		}
 
 		@Override