Add create session

Author: Maksim Timonin

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java

@@ -722,7 +722,7 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
 
                 sock = spi.openSocket(addr, timeoutHelper);
 
-                TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, spi);
+                TcpDiscoveryIoSession ses = createSession(sock);
 
                 openSock = true;
 
@@ -1176,7 +1176,7 @@ private void forceStopRead() throws InterruptedException {
                     + ":" + sockStream.sock.getPort());
 
                 try {
-                    TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, spi);
+                    TcpDiscoveryIoSession ses = createSession(sock);
 
                     assert sock.getKeepAlive() && sock.getTcpNoDelay() : "Socket wasn't configured properly:" +
                         " KeepAlive " + sock.getKeepAlive() +
@@ -1333,7 +1333,7 @@ private void setSocket(Socket sock, boolean clientAck) {
             synchronized (mux) {
                 this.sock = sock;
 
-                ses = new TcpDiscoveryIoSession(sock, spi);
+                ses = createSession(sock);
 
                 this.clientAck = clientAck;
 
@@ -1595,7 +1595,7 @@ public void cancel() {
                     clientAck = joinRes.get2();
 
                     Socket sock = sockStream.socket();
-                    TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, spi);
+                    TcpDiscoveryIoSession ses = createSession(sock);
 
                     if (isInterrupted())
                         throw new InterruptedException();

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java

@@ -874,7 +874,7 @@ private boolean pingNode(TcpDiscoveryNode node) {
 
                         openedSock = true;
 
-                        TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, spi);
+                        TcpDiscoveryIoSession ses = createSession(sock);
 
                         spi.writeMessage(ses, new TcpDiscoveryPingRequest(locNodeId, clientNodeId),
                             timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
@@ -1467,7 +1467,7 @@ else if (U.millisSinceNanos(joinStartNanos) > spi.joinTimeout)
                 long tsNanos = System.nanoTime();
 
                 sock = spi.openSocket(addr, timeoutHelper);
-                TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, spi);
+                TcpDiscoveryIoSession ses = createSession(sock);
 
                 openSock = true;
 
@@ -3520,7 +3520,7 @@ else if (log.isTraceEnabled())
                             try {
                                 sock = spi.openSocket(addr, timeoutHelper);
 
-                                ses = new TcpDiscoveryIoSession(sock, spi);
+                                ses = createSession(sock);
 
                                 openSock = true;
 
@@ -6706,7 +6706,7 @@ private class SocketReader extends IgniteSpiThread {
 
             this.sock = sock;
 
-            ses = new TcpDiscoveryIoSession(sock, spi);
+            ses = createSession(sock);
 
             setPriority(spi.threadPri);
         }
@@ -7734,7 +7734,7 @@ private ClientMessageWorker(Socket sock, UUID clientNodeId, IgniteLogger log) {
             this.sock = sock;
             this.clientNodeId = clientNodeId;
 
-            ses = new TcpDiscoveryIoSession(sock, spi);
+            ses = createSession(sock);
 
             lastMetricsUpdateMsgTimeNanos = System.nanoTime();
         }

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.spi.discovery.tcp;
 
 import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
@@ -439,6 +440,16 @@ protected static List<String> toOrderedList(Collection<InetSocketAddress> addrs)
         return res;
     }
 
+    /**
+     * Instantiates IO session for exchanging discovery messages with remote node.
+     *
+     * @param sock Socket to remote node.
+     * @return IO session for writing and reading {@link TcpDiscoveryAbstractMessage}.
+     */
+    TcpDiscoveryIoSession createSession(Socket sock) {
+        return new TcpDiscoveryIoSession(sock, spi);
+    }
+
     /**
      * @param msg Message.
      * @return Message logger.