[rdct] RDCT-1663. Connection related refactoring to prevent connection freezes

- Use dedicated connection flow thread on a backend side when processing a connection;
- Add check for a start receiving messages logic in SecureWireOverStreamTransport class to make sure wire still has valid state when start delivering messages;
- Unify logic to check for a connected state in ByteChannel and propagating state to the whole transport structure;
- Update port forwarding remove() method to call methods that require EDT on a correct thread;
- Minor cleanup in port forwarding code.


(cherry picked from commit 9dac873163af19e81ecd9e8def7e746e59d04ae4)

IJ-CR-146006

GitOrigin-RevId: 836ba1a369e8add127b6f92e79998ad04c7e3775

platform/core-api/src/com/intellij/openapi/client/ClientSessionsManager.kt

@@ -5,6 +5,7 @@ import com.intellij.codeWithMe.ClientId
 import com.intellij.openapi.Disposable
 import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.components.service
+import com.intellij.openapi.diagnostic.debug
 import com.intellij.openapi.diagnostic.logger
 import com.intellij.openapi.project.Project
 import com.intellij.openapi.util.Disposer
@@ -12,6 +13,7 @@ import org.jetbrains.annotations.ApiStatus
 import java.util.concurrent.ConcurrentHashMap
 
 private val LOG = logger<ClientSessionsManager<*>>()
+
 @ApiStatus.Experimental
 @ApiStatus.Internal
 open class ClientSessionsManager<T : ClientSession> {
@@ -77,12 +79,12 @@ open class ClientSessionsManager<T : ClientSession> {
     if (sessions.putIfAbsent(clientId, session) != null) {
       LOG.error("Session $session with such clientId is already registered")
     }
-    LOG.debug("Session added '$session'")
+    LOG.debug { "Session added '$session'" }
 
     Disposer.register(disposable, session)
     Disposer.register(disposable) {
       sessions.remove(clientId)
-      LOG.debug("Session removed '$clientId'")
+      LOG.debug { "Session removed '$clientId'" }
     }
   }