Split out TcpMuxer

app/src/main/java/io/bimmergestalt/bcl/multiplex/BclProxyServer.kt

@@ -0,0 +1,78 @@
+package io.bimmergestalt.bcl.multiplex
+
+import io.bimmergestalt.bcl.client.ProxyConnectionGrantor
+import org.tinylog.kotlin.Logger
+import java.io.IOException
+import java.io.OutputStream
+import java.net.InetSocketAddress
+import java.nio.ByteBuffer
+import java.nio.channels.CancelledKeyException
+import java.nio.channels.SelectionKey
+import java.nio.channels.Selector
+import java.nio.channels.ServerSocketChannel
+import java.nio.channels.SocketChannel
+
+class BclProxyServer(val listenPort: Short, val connection: ProxyConnectionGrantor) {
+    val selector = Selector.open()
+    val serverSocket = ServerSocketChannel.open()
+    val muxer = TcpMuxer(selector)
+
+    fun listen() {
+        serverSocket.configureBlocking(false)
+        serverSocket.register(selector, SelectionKey.OP_ACCEPT)
+        serverSocket.bind(InetSocketAddress(listenPort.toInt()))
+    }
+
+    fun run() {
+        while (serverSocket.isOpen) {
+            selector.select(10000)
+            val readyKeys = selector.selectedKeys()
+            readyKeys.forEach { key ->
+                try {
+                    if (key.isAcceptable) {
+                        // new client connection
+                        val socket: SocketChannel = serverSocket.accept()
+                        val bclProxyConnection = connection.openConnection(SocketChannelOutputStream(socket))
+                        muxer.addChannel(socket, bclProxyConnection)
+                    }
+                } catch (_: CancelledKeyException) {}
+            }
+            readyKeys.clear()
+
+            muxer.process()
+        }
+        shutdown()
+    }
+
+    fun shutdown() {
+        try {
+            serverSocket.close()
+        } catch (e: IOException) {
+            Logger.warn(e) { "IOException while shutting down BclProxy ServerSocket" }
+        }
+        muxer.shutdown()
+    }
+}
+
+/**
+ * Masquerades a SocketChannel as an OutputStream
+ */
+class SocketChannelOutputStream(private val channel: SocketChannel): OutputStream() {
+    override fun write(b: Int) {
+        channel.write(ByteBuffer.wrap(byteArrayOf(b.toByte())))
+    }
+
+    override fun write(b: ByteArray?) {
+        b ?: return
+        channel.write(ByteBuffer.wrap(b))
+    }
+
+    override fun write(b: ByteArray?, off: Int, len: Int) {
+        b ?: return
+        channel.write(ByteBuffer.wrap(b, off, len))
+    }
+
+    override fun close() {
+        channel.close()
+    }
+}

app/src/main/java/io/bimmergestalt/bcl/multiplex/TcpMuxer.kt

@@ -0,0 +1,66 @@
+package io.bimmergestalt.bcl.multiplex
+
+import io.bimmergestalt.bcl.protocols.ProxyClientConnection
+import org.tinylog.kotlin.Logger
+import java.io.IOException
+import java.nio.ByteBuffer
+import java.nio.channels.CancelledKeyException
+import java.nio.channels.SelectionKey
+import java.nio.channels.Selector
+import java.nio.channels.SocketChannel
+
+class TcpMuxer(val selector: Selector) {
+    val connections = HashMap<SocketChannel, ProxyClientConnection>()
+    fun addChannel(channel: SocketChannel, muxStream: ProxyClientConnection) {
+        channel.configureBlocking(false)
+        channel.socket().tcpNoDelay = true
+        channel.register(selector, SelectionKey.OP_READ)
+        connections[channel] = muxStream
+    }
+
+    fun process() {
+        val inputBuffer = ByteBuffer.allocate(4000)
+
+        selector.selectNow()
+        val readyKeys = selector.selectedKeys()
+        readyKeys.forEach { key ->
+            try {
+                if (key.isReadable) {
+                    // new data from the client
+                    val channel = key.channel()
+                    if (channel is SocketChannel) {
+                        val len = channel.read(inputBuffer)
+//                            Logger.debug { "Read $len bytes from client socket" }
+                        if (len > 0) {
+                            try {
+                                connections[channel]?.toTunnel?.write(inputBuffer.array(), 0, len)
+                            } catch (e: IOException) {
+                                Logger.warn(e) { "IOException while writing to BclConnection from client" }
+                                channel.close()
+                                connections.remove(channel)?.close()
+                            }
+                        } else if (len == -1) {
+                            key.cancel()
+                            connections.remove(channel)?.close()
+                        }
+                        inputBuffer.clear()
+                    }
+                }
+            } catch (_: CancelledKeyException) {}
+        }
+        readyKeys.clear()
+    }
+
+    fun shutdown() {
+        synchronized(connections) {
+            connections.keys.forEach {
+                try {
+                    it.close()
+                } catch (e: IOException) {
+                    Logger.warn(e) { "IOException while shutting down muxer client" }
+                }
+            }
+            connections.clear()
+        }
+    }
+}

app/src/main/java/io/bimmergestalt/bcl/protocols/TcpProxyProtocol.kt

@@ -1,6 +1,6 @@
 package io.bimmergestalt.bcl.protocols
 
-import io.bimmergestalt.bcl.BclProxyServer
+import io.bimmergestalt.bcl.multiplex.BclProxyServer
 import io.bimmergestalt.bcl.ConnectionState
 import io.bimmergestalt.bcl.MutableConnectionState
 import io.bimmergestalt.bcl.client.ProxyConnectionGrantor