moving read operations to a new thread and selector

src/Client.kt

@@ -4,5 +4,10 @@ import java.nio.charset.StandardCharsets
 fun main(args: Array<String>) {
  val socket = Socket("localhost", 10301)
  val input = socket.getOutputStream()
- input.write("Hello Server".toByteArray(StandardCharsets.UTF_8))
+ input.write(("Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message " +
+ "Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message " +
+ "Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message " +
+ "Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message " +
+ "Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message " +
+ "Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer messageHello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message Hello Server this is a longer message ").toByteArray(StandardCharsets.UTF_8))
 }

src/Server.kt

@@ -8,6 +8,7 @@ fun main(args: Array<String>) {
  val server = Server(10301)
 
  val serverThread = Thread(server)
+ serverThread.name = "NIO Server Acceptor"
  Runtime.getRuntime().addShutdownHook(Thread {
  serverThread.interrupt()
  })
@@ -20,17 +21,24 @@ class Server(port: Int) : Runnable {
  val socketChannel: ServerSocketChannel = ServerSocketChannel.open()
  val selector: Selector
 
+ val readerThread: Thread
+ val serverReader: ServerReader = ServerReader()
+
  init {
  socketChannel.socket().bind(InetSocketAddress(port))
  socketChannel.configureBlocking(false)
 
  selector = Selector.open()
  socketChannel.register(selector, SelectionKey.OP_ACCEPT)
+
+ readerThread = Thread(serverReader)
+ readerThread.name = "NIO Server Reader"
+ readerThread.start()
  }
 
  override fun run() {
 
- println("Server socket listening on ${socketChannel.socket().localPort}")
+ println("[${Thread.currentThread().name}] - Server socket listening on ${socketChannel.socket().localPort}")
 
  try {
 
@@ -45,21 +53,18 @@ class Server(port: Int) : Runnable {
  accept(key)
  }
 
- if (key.isReadable) {
- val clientId = key.attachment() as String
-
- }
-
  iter.remove()
  }
  }
 
- println("Socket closing gracefully")
+ println("[${Thread.currentThread().name}] - Socket closing gracefully")
  socketChannel.close()
 
  } catch (e: IOException) {
- System.err.println("Closing down socket - ${e.printStackTrace()}")
+ System.err.println("[${Thread.currentThread().name}] - Closing down socket - ${e.printStackTrace()}")
  }
+
+ readerThread.interrupt()
  }
 
  private fun accept(selectionKey: SelectionKey) {
@@ -69,9 +74,9 @@ class Server(port: Int) : Runnable {
  clientSocket.configureBlocking(false)
 
  val address = clientSocket.socket().inetAddress.toString()
- val port = clientSocket.socket().localPort
+ val port = clientSocket.socket().port
  val clientId = "$address:$port"
 
- clientSocket.register(selector, SelectionKey.OP_READ, clientId)
+ clientSocket.register(serverReader.selector, SelectionKey.OP_READ, clientId)
  }
 }

src/ServerReader.kt

@@ -0,0 +1,50 @@
+import java.nio.ByteBuffer
+import java.nio.channels.SelectionKey
+import java.nio.channels.Selector
+import java.nio.channels.SocketChannel
+import java.nio.charset.StandardCharsets
+
+class ServerReader : Runnable {
+
+ val selector: Selector = Selector.open()
+
+ override fun run() {
+
+ println("[${Thread.currentThread().name}] - Server reader thread is now running")
+ while (!Thread.interrupted()) {
+
+ selector.select(1000)
+
+ val iter = selector.selectedKeys().iterator()
+ iter.forEach { key ->
+
+ if (key.isReadable) {
+ read(key)
+ }
+
+ iter.remove()
+ }
+ }
+
+ println("[${Thread.currentThread().name}] - Server reader thread stopping")
+ }
+
+ private fun read(selectionKey: SelectionKey) {
+ val clientId = selectionKey.attachment() as String
+
+ val socket = selectionKey.channel() as SocketChannel
+
+ //We know this is more than what we are sending - we will deal with multiple buffers later
+ val buffer = ByteBuffer.allocate(512)
+
+ val bytesRead = socket.read(buffer)
+ if (bytesRead > 0) {
+ println("---- [${Thread.currentThread().name}] - Reading from $clientId")
+
+ println("[${Thread.currentThread().name}] - $bytesRead bytes from $clientId")
+
+ val message = String(buffer.array().slice(0 until bytesRead).toByteArray(), StandardCharsets.UTF_8)
+ println("[${Thread.currentThread().name}] - PAYLOAD { $message }")
+ }
+ }
+}