need to override socket input/output stream

Author: ekoby

ziti/src/integrationTest/kotlin/org/openziti/net/ConnectionTests.kt

@@ -167,7 +167,7 @@ class ConnectionTests: BaseTest() {
     }
 
     @Test
-    fun `test socket-connect-read-timeout`() = runTest(timeout = 10.seconds) {
+    fun `test socket-connect-read-timeout`() = runTest(timeout = 1000.seconds) {
         val greeting = "Hello from Ziti".toByteArray()
         val s = assertDoesNotThrow {
             ztx.serviceUpdates().filter { it.service.name == service }.first().service
@@ -195,6 +195,7 @@ class ConnectionTests: BaseTest() {
             }
 
             ztx.connect(hostname, port).use { clt ->
+                assertTrue(clt.isConnected)
                 val buf = ByteArray(1024)
                 clt.soTimeout = 500
                 val input = clt.getInputStream()

ziti/src/main/kotlin/org/openziti/impl/ZitiContextImpl.kt

@@ -214,16 +214,17 @@ internal class ZitiContextImpl(internal val id: Identity, enabled: Boolean) : Zi
 
     override fun connect(host: String, port: Int): Socket {
         checkEnabled()
-        val ch = open()
-        runCatching {
+        return runCatching {
+            val ch = open()
             ch.connect(InetSocketAddress.createUnresolved(host, port)).get(10, TimeUnit.SECONDS)
-        }.onFailure {
+            d { "connected: ${ch.remoteAddress}" }
+            AsychChannelSocket(ch)
+        }.getOrElse {
             when (it) {
                 is ExecutionException -> throw it.cause ?: it
                 else -> throw it
             }
         }
-        return AsychChannelSocket(ch)
     }
 
     fun start(): Job = launch {

ziti/src/main/kotlin/org/openziti/net/nio/AsychChannelSocket.kt

@@ -16,11 +16,12 @@
 
 package org.openziti.net.nio
 
+import java.io.*
 import java.net.InetAddress
 import java.net.InetSocketAddress
 import java.net.Socket
+import java.net.SocketException
 import java.nio.channels.AsynchronousSocketChannel
-import javax.net.SocketFactory
 
 /**
  * Socket wrapper over AsynchronousSocketChannel.
@@ -32,20 +33,6 @@ import javax.net.SocketFactory
 internal
 class AsychChannelSocket(internal val impl: AsyncSocketImpl = AsyncSocketImpl()): Socket(impl) {
 
-    class Factory: SocketFactory() {
-        override fun createSocket(): Socket  = AsychChannelSocket()
-        override fun createSocket(host: String?, port: Int): Socket  = AsychChannelSocket(host, port)
-        override fun createSocket(host: InetAddress?, port: Int): Socket  = AsychChannelSocket(host, port)
-
-        override fun createSocket(host: String?, port: Int, localHost: InetAddress?, localPort: Int): Socket  {
-            TODO("Not yet implemented")
-        }
-
-        override fun createSocket(address: InetAddress?, port: Int, localAddress: InetAddress?, localPort: Int): Socket {
-            TODO("Not yet implemented")
-        }
-    }
-
     constructor(ch: AsynchronousSocketChannel): this(AsyncSocketImpl(ch))
 
     constructor(host: String?, port: Int) : this(InetAddress.getByName(host), port)
@@ -64,4 +51,29 @@ class AsychChannelSocket(internal val impl: AsyncSocketImpl = AsyncSocketImpl())
         impl.channel.close()
     }
 
+    override fun getInputStream(): InputStream {
+        if (isClosed())
+            throw SocketException("Socket is closed")
+        if (!isConnected)
+            throw SocketException("Socket is not connected")
+
+        return object : FilterInputStream(impl.inputStream) {
+            override fun close() {
+                this@AsychChannelSocket.close()
+            }
+        }
+    }
+
+    override fun getOutputStream(): OutputStream {
+        if (isClosed())
+            throw SocketException("Socket is closed")
+        if (!isConnected)
+            throw SocketException("Socket is not connected")
+
+        return object : FilterOutputStream(impl.outputStream) {
+            override fun close() {
+                this@AsychChannelSocket.close()
+            }
+        }
+    }
 }