对项目进行进一步拆分

settings.gradle

@@ -1,3 +1,3 @@
 rootProject.name = 'TursomServer'
-include 'web', 'aop', 'database', 'database:database-async', 'socket', 'utils', 'utils:xml', 'utils:async-http', 'web:netty-web'
-
+include 'web', 'aop', 'database', 'database:database-async', 'utils', 'utils:xml', 'utils:async-http', 'web:netty-web'
+include 'socket', 'socket:socket-async'

socket/build.gradle

@@ -1,6 +1,3 @@
 dependencies {
     implementation project(":")
-
-    // kotlin 协程
-    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1'
 }

socket/socket-async/build.gradle

@@ -0,0 +1,7 @@
+dependencies {
+    implementation project(":")
+    implementation project(":socket")
+
+    // kotlin 协程
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1'
+}

socket/socket-async/src/main/kotlin/cn/tursom/socket/AsyncAioSocket.kt

@@ -0,0 +1,72 @@
+package cn.tursom.socket
+
+import java.net.SocketAddress
+import java.nio.ByteBuffer
+import java.nio.channels.AsynchronousSocketChannel
+import java.nio.channels.CompletionHandler
+import java.util.concurrent.TimeUnit
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlin.coroutines.suspendCoroutine
+
+@Suppress("MemberVisibilityCanBePrivate")
+open class AsyncAioSocket(val socketChannel: AsynchronousSocketChannel) : AsyncSocket {
+    val address: SocketAddress get() = socketChannel.remoteAddress
+
+    override suspend fun write(buffer: ByteBuffer, timeout: Long): Int {
+        return suspendCoroutine { cont ->
+            this.socketChannel.write(buffer, timeout, TimeUnit.MILLISECONDS, cont, awaitHandler)
+        }
+    }
+
+    override suspend fun read(buffer: ByteBuffer, timeout: Long): Int {
+        return suspendCoroutine { cont ->
+            this.socketChannel.read(buffer, timeout, TimeUnit.MILLISECONDS, cont, awaitHandler)
+        }
+    }
+
+    override suspend fun write(buffer: Array<out ByteBuffer>, timeout: Long): Long {
+        return suspendCoroutine { cont ->
+            this.socketChannel.write(buffer, 0, buffer.size, timeout, TimeUnit.MILLISECONDS, cont, awaitLongHandler)
+        }
+    }
+
+    override suspend fun read(buffer: Array<out ByteBuffer>, timeout: Long): Long {
+        return suspendCoroutine { cont ->
+            this.socketChannel.read(buffer, 0, buffer.size, timeout, TimeUnit.MILLISECONDS, cont, awaitLongHandler)
+        }
+    }
+
+    override fun close() {
+        socketChannel.close()
+    }
+
+    companion object {
+        const val defaultTimeout = 60_000L
+
+        @JvmStatic
+        private val awaitHandler =
+            object : CompletionHandler<Int, Continuation<Int>> {
+                override fun completed(result: Int, attachment: Continuation<Int>) {
+                    attachment.resume(result)
+                }
+
+                override fun failed(exc: Throwable, attachment: Continuation<Int>) {
+                    attachment.resumeWithException(exc)
+                }
+            }
+
+        @JvmStatic
+        private val awaitLongHandler =
+            object : CompletionHandler<Long, Continuation<Long>> {
+                override fun completed(result: Long, attachment: Continuation<Long>) {
+                    attachment.resume(result)
+                }
+
+                override fun failed(exc: Throwable, attachment: Continuation<Long>) {
+                    attachment.resumeWithException(exc)
+                }
+            }
+    }
+}

socket/socket-async/src/main/kotlin/cn/tursom/socket/AsyncClient.kt

@@ -1,7 +1,5 @@
-package cn.tursom.socket.client
+package cn.tursom.socket
 
-import cn.tursom.socket.AsyncCachedAioSocket
-import cn.tursom.socket.AsyncAioSocket
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import java.net.InetSocketAddress
@@ -35,7 +33,9 @@ object AsyncClient {
 	
 	suspend fun connect(socketChannel: AsynchronousSocketChannel, host: String, port: Int): AsyncAioSocket {
 		suspendCoroutine<Void?> { cont ->
-			socketChannel.connect(InetSocketAddress(host, port) as SocketAddress, cont, handler)
+			socketChannel.connect(InetSocketAddress(host, port) as SocketAddress, cont,
+                handler
+            )
 		}
 		return AsyncAioSocket(socketChannel)
 	}
@@ -47,7 +47,9 @@ object AsyncClient {
 	
 	suspend fun connectCached(socketChannel: AsynchronousSocketChannel, host: String, port: Int): AsyncAioSocket {
 		suspendCoroutine<Void?> { cont ->
-			socketChannel.connect(InetSocketAddress(host, port) as SocketAddress, cont, handler)
+			socketChannel.connect(InetSocketAddress(host, port) as SocketAddress, cont,
+                handler
+            )
 		}
 		return AsyncCachedAioSocket(socketChannel)
 	}

socket/socket-async/src/main/kotlin/cn/tursom/socket/AsyncNioClient.kt

@@ -1,6 +1,5 @@
-package cn.tursom.socket.client
+package cn.tursom.socket
 
-import cn.tursom.socket.AsyncNioSocket
 import cn.tursom.socket.niothread.WorkerLoopNioThread
 import java.net.InetSocketAddress
 import java.nio.channels.SelectionKey

socket/socket-async/src/main/kotlin/cn/tursom/socket/AsyncNioSocket.kt

@@ -0,0 +1,259 @@
+package cn.tursom.socket
+
+import cn.tursom.socket.niothread.INioThread
+import cn.tursom.core.timer.TimerTask
+import cn.tursom.core.timer.WheelTimer
+import java.nio.ByteBuffer
+import java.nio.channels.SelectionKey
+import java.nio.channels.SocketChannel
+import java.util.concurrent.TimeoutException
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlin.coroutines.suspendCoroutine
+
+/**
+ * 利用 SelectionKey 的 attachment 进行状态的传输
+ * 导致该类无法利用 SelectionKey 的 attachment
+ * 但是对于一般的应用而言是足够使用的
+ */
+class AsyncNioSocket(override val key: SelectionKey, override val nioThread: INioThread) : IAsyncNioSocket {
+    override val channel: SocketChannel = key.channel() as SocketChannel
+
+    override suspend fun read(buffer: ByteBuffer): Int {
+        if (buffer.remaining() == 0) return -1
+        return try {
+            suspendCoroutine {
+                key.attach(SingleContext(buffer, it))
+                readMode()
+                nioThread.wakeup()
+            }
+        } catch (e: Exception) {
+            waitMode()
+            throw RuntimeException(e)
+        }
+    }
+
+    override suspend fun read(buffer: Array<out ByteBuffer>): Long {
+        if (buffer.size == 0) return -1
+        return try {
+            suspendCoroutine {
+                key.attach(MultiContext(buffer, it))
+                readMode()
+                nioThread.wakeup()
+            }
+        } catch (e: Exception) {
+            waitMode()
+            throw RuntimeException(e)
+        }
+    }
+
+    override suspend fun write(buffer: ByteBuffer): Int {
+        if (buffer.remaining() == 0) return -1
+        return try {
+            suspendCoroutine {
+                key.attach(SingleContext(buffer, it))
+                writeMode()
+                nioThread.wakeup()
+            }
+        } catch (e: Exception) {
+            waitMode()
+            throw Exception(e)
+        }
+    }
+
+    override suspend fun write(buffer: Array<out ByteBuffer>): Long {
+        if (buffer.isEmpty()) return -1
+        return try {
+            suspendCoroutine {
+                key.attach(MultiContext(buffer, it))
+                writeMode()
+                nioThread.wakeup()
+            }
+        } catch (e: Exception) {
+            waitMode()
+            throw Exception(e)
+        }
+    }
+
+    override suspend fun read(buffer: ByteBuffer, timeout: Long): Int {
+        if (timeout <= 0) return read(buffer)
+        if (buffer.remaining() == 0) return -1
+        return try {
+            val result: Int = suspendCoroutine {
+                key.attach(
+                    SingleContext(
+                        buffer,
+                        it,
+                        timer.exec(timeout) {
+                            try {
+                                it.resumeWithException(TimeoutException())
+                            } catch (e: Exception) {
+                            }
+                        })
+                )
+                readMode()
+                nioThread.wakeup()
+            }
+            result
+        } catch (e: Exception) {
+            waitMode()
+            throw RuntimeException(e)
+        }
+    }
+
+    override suspend fun read(buffer: Array<out ByteBuffer>, timeout: Long): Long {
+        if (timeout <= 0) return read(buffer)
+        if (buffer.isEmpty()) return -1
+        return try {
+            val result: Long = suspendCoroutine {
+                key.attach(
+                    MultiContext(
+                        buffer,
+                        it,
+                        timer.exec(timeout) {
+                            try {
+                                it.resumeWithException(TimeoutException())
+                            } catch (e: Exception) {
+                            }
+                        })
+                )
+                readMode()
+                nioThread.wakeup()
+            }
+            result
+        } catch (e: Exception) {
+            waitMode()
+            throw Exception(e)
+        }
+    }
+
+    override suspend fun write(buffer: ByteBuffer, timeout: Long): Int {
+        if (timeout <= 0) return write(buffer)
+        if (buffer.remaining() == 0) return -1
+        return try {
+            val result: Int = suspendCoroutine {
+                key.attach(
+                    SingleContext(
+                        buffer,
+                        it,
+                        timer.exec(timeout) {
+                            try {
+                                it.resumeWithException(TimeoutException())
+                            } catch (e: Exception) {
+                            }
+                        })
+                )
+                writeMode()
+                nioThread.wakeup()
+            }
+            result
+        } catch (e: Exception) {
+            waitMode()
+            throw Exception(e)
+        }
+    }
+
+    override suspend fun write(buffer: Array<out ByteBuffer>, timeout: Long): Long {
+        if (timeout <= 0) return write(buffer)
+        if (buffer.isEmpty()) return -1
+        return try {
+            val result: Long = suspendCoroutine {
+                key.attach(
+                    MultiContext(
+                        buffer,
+                        it,
+                        timer.exec(timeout) {
+                            try {
+                                it.resumeWithException(TimeoutException())
+                            } catch (e: Exception) {
+                            }
+                        })
+                )
+                writeMode()
+                nioThread.wakeup()
+            }
+            result
+        } catch (e: Exception) {
+            waitMode()
+            throw Exception(e)
+        }
+    }
+
+    override fun close() {
+        nioThread.execute {
+            channel.close()
+            key.cancel()
+        }
+    }
+
+    interface Context {
+        val cont: Continuation<*>
+        val timeoutTask: TimerTask? get() = null
+    }
+
+    class SingleContext(
+        val buffer: ByteBuffer,
+        override val cont: Continuation<Int>,
+        override val timeoutTask: TimerTask? = null
+    ) : Context
+
+    class MultiContext(
+        val buffer: Array<out ByteBuffer>,
+        override val cont: Continuation<Long>,
+        override val timeoutTask: TimerTask? = null
+    ) : Context
+
+    companion object {
+        val nioSocketProtocol = object : INioProtocol {
+            override fun handleConnect(key: SelectionKey, nioThread: INioThread) {}
+
+            override fun handleRead(key: SelectionKey, nioThread: INioThread) {
+                key.interestOps(0)
+                val context = key.attachment() as Context
+                context.timeoutTask?.cancel()
+                if (context is SingleContext) {
+                    val channel = key.channel() as SocketChannel
+                    val readSize = channel.read(context.buffer)
+                    context.cont.resume(readSize)
+                } else {
+                    context as MultiContext
+                    val channel = key.channel() as SocketChannel
+                    val readSize = channel.read(context.buffer)
+                    context.cont.resume(readSize)
+                }
+            }
+
+            override fun handleWrite(key: SelectionKey, nioThread: INioThread) {
+                key.interestOps(0)
+                val context = key.attachment() as Context
+                context.timeoutTask?.cancel()
+                if (context is SingleContext) {
+                    val channel = key.channel() as SocketChannel
+                    val readSize = channel.write(context.buffer)
+                    context.cont.resume(readSize)
+                } else {
+                    context as MultiContext
+                    val channel = key.channel() as SocketChannel
+                    val readSize = channel.write(context.buffer)
+                    context.cont.resume(readSize)
+                }
+            }
+
+            override fun exceptionCause(key: SelectionKey, nioThread: INioThread, e: Throwable) {
+                key.interestOps(0)
+                val context = key.attachment() as Context?
+                if (context != null)
+                    context.cont.resumeWithException(e)
+                else {
+                    key.cancel()
+                    key.channel().close()
+                    e.printStackTrace()
+                }
+            }
+        }
+
+        //val timer = StaticWheelTimer.timer
+        val timer = WheelTimer.timer
+    }
+}

socket/socket-async/src/main/kotlin/cn/tursom/socket/server/AsyncGroupNioServer.kt

@@ -1,9 +1,8 @@
-package cn.tursom.socket.server.nio
+package cn.tursom.socket.server
 
 import cn.tursom.socket.AsyncNioSocket
 import cn.tursom.socket.INioProtocol
 import cn.tursom.socket.niothread.INioThread
-import cn.tursom.socket.server.ISocketServer
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import java.nio.channels.SelectionKey
@@ -18,20 +17,25 @@ class AsyncGroupNioServer(
 	val threads: Int = Runtime.getRuntime().availableProcessors(),
 	backlog: Int = 50,
 	val handler: suspend AsyncNioSocket.() -> Unit
-) : ISocketServer by GroupNioServer(port, threads, object : INioProtocol by AsyncNioSocket.nioSocketProtocol {
-	override fun handleConnect(key: SelectionKey, nioThread: INioThread) {
-		GlobalScope.launch {
-			val socket = AsyncNioSocket(key, nioThread)
-			try {
-				socket.handler()
-			} catch (e: Exception) {
-				e.printStackTrace()
-			} finally {
+) : ISocketServer by GroupNioServer(
+	port,
+	threads,
+	object : INioProtocol by AsyncNioSocket.nioSocketProtocol {
+		override fun handleConnect(key: SelectionKey, nioThread: INioThread) {
+			GlobalScope.launch {
+				val socket = AsyncNioSocket(key, nioThread)
 				try {
-					nioThread.execute { socket.close() }
+					socket.handler()
 				} catch (e: Exception) {
+					e.printStackTrace()
+				} finally {
+					try {
+						nioThread.execute { socket.close() }
+					} catch (e: Exception) {
+					}
 				}
 			}
 		}
-	}
-}, backlog)
+	},
+	backlog
+)

socket/socket-async/src/main/kotlin/cn/tursom/socket/server/AsyncNioServer.kt

@@ -1,9 +1,8 @@
-package cn.tursom.socket.server.nio
+package cn.tursom.socket.server
 
 import cn.tursom.socket.AsyncNioSocket
 import cn.tursom.socket.INioProtocol
 import cn.tursom.socket.niothread.INioThread
-import cn.tursom.socket.server.ISocketServer
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import java.nio.channels.SelectionKey
@@ -18,21 +17,21 @@ class AsyncNioServer(
 	backlog: Int = 50,
 	val handler: suspend AsyncNioSocket.() -> Unit
 ) : ISocketServer by NioServer(port, object : INioProtocol by AsyncNioSocket.nioSocketProtocol {
-	override fun handleConnect(key: SelectionKey, nioThread: INioThread) {
-		GlobalScope.launch {
-			val socket = AsyncNioSocket(key, nioThread)
-			try {
-				socket.handler()
-			} catch (e: Exception) {
-				e.printStackTrace()
-			} finally {
-				try {
-					socket.close()
-				} catch (e: Exception) {
-				}
-			}
-		}
-	}
+    override fun handleConnect(key: SelectionKey, nioThread: INioThread) {
+        GlobalScope.launch {
+            val socket = AsyncNioSocket(key, nioThread)
+            try {
+                socket.handler()
+            } catch (e: Exception) {
+                e.printStackTrace()
+            } finally {
+                try {
+                    socket.close()
+                } catch (e: Exception) {
+                }
+            }
+        }
+    }
 }, backlog) {
 	/**
 	 * 次要构造方法，为使用Spring的同学们准备的

socket/src/main/kotlin/cn/tursom/socket/AsyncAioSocket.kt

@@ -1,297 +0,0 @@
-package cn.tursom.socket
-
-import cn.tursom.core.*
-import cn.tursom.socket.client.AsyncClient
-import cn.tursom.socket.server.async.AsyncSocketServer
-import cn.tursom.core.bytebuffer.HeapByteBuffer
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
-import java.io.ByteArrayOutputStream
-import java.io.OutputStream
-import java.nio.ByteBuffer
-import java.nio.channels.AsynchronousSocketChannel
-import java.nio.channels.CompletionHandler
-import java.nio.channels.InterruptedByTimeoutException
-import java.util.concurrent.TimeUnit
-import kotlin.coroutines.Continuation
-import kotlin.coroutines.resume
-import kotlin.coroutines.resumeWithException
-import kotlin.coroutines.suspendCoroutine
-
-open class AsyncAioSocket(private val socketChannel: AsynchronousSocketChannel) : AsyncSocket {
-	val address get() = socketChannel.remoteAddress
-
-	fun cached() = AsyncCachedAioSocket(socketChannel)
-
-	override suspend fun write(buffer: ByteBuffer, timeout: Long): Int {
-		return suspendCoroutine { cont ->
-			this.socketChannel.write(buffer, timeout, TimeUnit.MILLISECONDS, cont, awaitHandler)
-		}
-	}
-
-	override suspend fun read(buffer: ByteBuffer, timeout: Long): Int {
-		return suspendCoroutine { cont ->
-			this.socketChannel.read(buffer, timeout, TimeUnit.MILLISECONDS, cont, awaitHandler)
-		}
-	}
-
-	override suspend fun write(buffer: Array<out ByteBuffer>, timeout: Long): Long {
-		return suspendCoroutine { cont ->
-			this.socketChannel.write(buffer, 0, buffer.size, timeout, TimeUnit.MILLISECONDS, cont, awaitLongHandler)
-		}
-	}
-
-	override suspend fun read(buffer: Array<out ByteBuffer>, timeout: Long): Long {
-		return suspendCoroutine { cont ->
-			this.socketChannel.read(buffer, 0, buffer.size, timeout, TimeUnit.MILLISECONDS, cont, awaitLongHandler)
-		}
-	}
-
-	override fun close() {
-		socketChannel.close()
-	}
-
-	fun reset() {
-		val field = socketChannel.javaClass.getDeclaredField("readKilled")
-		field.isAccessible = true
-		field.set(socketChannel, true)
-	}
-
-	companion object {
-		const val defaultTimeout = 60_000L
-
-		@JvmStatic
-		private val awaitHandler =
-			object : CompletionHandler<Int, Continuation<Int>> {
-				override fun completed(result: Int, attachment: Continuation<Int>) {
-					attachment.resume(result)
-				}
-
-				override fun failed(exc: Throwable, attachment: Continuation<Int>) {
-					attachment.resumeWithException(exc)
-				}
-			}
-
-		@JvmStatic
-		private val awaitLongHandler =
-			object : CompletionHandler<Long, Continuation<Long>> {
-				override fun completed(result: Long, attachment: Continuation<Long>) {
-					attachment.resume(result)
-				}
-
-				override fun failed(exc: Throwable, attachment: Continuation<Long>) {
-					attachment.resumeWithException(exc)
-				}
-			}
-	}
-}
-
-@Suppress("unused")
-suspend inline fun AsyncAioSocket.write(message: String, timeout: Long = 0L) =
-	write(ByteBuffer.wrap(message.toByteArray()), timeout)
-
-suspend inline fun AsyncAioSocket.recvStr(buffer: ByteBuffer, timeout: Long = 0L): String {
-	//readBuffer.clear()
-	read(buffer, timeout)
-	return String(buffer.array(), buffer.arrayOffset(), buffer.position())
-}
-
-
-suspend inline fun <T : OutputStream> AsyncAioSocket.recv(
-	outputStream: T,
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout,
-	buffer: ByteBuffer = ByteBuffer.allocate(1024)
-): T {
-	buffer.clear()
-
-	try {
-		if (read(buffer, firstTimeout) <= 0) throw InterruptedByTimeoutException()
-		@Suppress("BlockingMethodInNonBlockingContext")
-		outputStream.write(buffer.array(), buffer.arrayOffset(), buffer.position())
-		buffer.clear()
-
-		while (read(buffer, readTimeout) > 0) {
-			@Suppress("BlockingMethodInNonBlockingContext")
-			outputStream.write(buffer.array(), buffer.arrayOffset(), buffer.position())
-			buffer.clear()
-		}
-	} catch (e: InterruptedByTimeoutException) {
-	}
-
-	return outputStream
-}
-
-suspend inline fun AsyncAioSocket.recv(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout,
-	buffer: ByteBuffer = ByteBuffer.allocate(1024)
-): ByteArray {
-	buffer.clear()
-	val byteStream = ByteArrayOutputStream()
-	recv(byteStream, readTimeout, firstTimeout, buffer)
-	return byteStream.toByteArray()
-}
-
-suspend inline fun AsyncAioSocket.recvStr(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout,
-	buffer: ByteBuffer = ByteBuffer.allocate(1024)
-): String {
-	buffer.clear()
-	val byteStream = ByteArrayOutputStream()
-	recv(byteStream, readTimeout, firstTimeout, buffer)
-	return String(byteStream.buf, 0, byteStream.count)
-}
-
-suspend inline fun AsyncAioSocket.recvInt(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout,
-	buffer: ByteBuffer = ByteBuffer.allocate(4)
-): Int {
-	buffer.clear().limit(4)
-	var readSize = read(buffer, firstTimeout)
-	while (readSize < 8) {
-		readSize += read(buffer, readTimeout)
-	}
-	return buffer.array().toInt(buffer.arrayOffset())
-}
-
-suspend inline fun AsyncAioSocket.recvLong(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout,
-	buffer: ByteBuffer = ByteBuffer.allocate(8)
-): Long {
-	buffer.clear().limit(8)
-	var readSize = read(buffer, firstTimeout)
-	while (readSize < 8) {
-		readSize += read(buffer, readTimeout)
-	}
-	return buffer.array().toLong(buffer.arrayOffset())
-}
-
-@Suppress("UNCHECKED_CAST")
-suspend inline fun <T> AsyncAioSocket.unSerializeObject(
-	buffer: ByteBuffer = ByteBuffer.allocate(1024),
-	readTimeout: Long = 100L,
-	firstTimeout: Long = AsyncAioSocket.defaultTimeout
-): T? {
-	return recv(ByteArrayOutputStream(), readTimeout, firstTimeout, buffer).let { unSerialize(
-        it.buf,
-        0,
-        it.count
-    ) as T? }
-}
-
-suspend inline fun AsyncAioSocket.send(message: ByteArray?, offset: Int = 0, size: Int = message?.size ?: 0) {
-	write(HeapByteBuffer.wrap(message ?: return, offset, size))
-}
-
-suspend inline fun AsyncAioSocket.send(message: String?) {
-	send((message ?: return).toByteArray())
-}
-
-suspend inline fun AsyncAioSocket.send(message: Int, buffer: ByteArray = ByteArray(4)) {
-	buffer.put(message)
-	send(buffer)
-}
-
-suspend inline fun AsyncAioSocket.send(message: Long, buffer: ByteArray = ByteArray(8)) {
-	buffer.put(message)
-	send(buffer)
-}
-
-suspend fun AsyncAioSocket.sendObject(obj: Any?): Int {
-	val byteArrayOutputStream = ByteArrayOutputStream()
-    serialize(byteArrayOutputStream, obj ?: return -1)
-	send(byteArrayOutputStream.buf, 0, byteArrayOutputStream.count)
-	return byteArrayOutputStream.count
-}
-
-inline fun <T> AsyncAioSocket.use(crossinline block: suspend AsyncAioSocket.() -> T): T {
-	var exception: Throwable? = null
-	try {
-		return runBlocking { block() }
-	} catch (e: Throwable) {
-		exception = e
-		throw e
-	} finally {
-		when (exception) {
-			null -> close()
-			else -> try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-}
-
-inline infix fun AsyncAioSocket.useNonBlock(crossinline block: suspend AsyncAioSocket.() -> Unit) =
-	GlobalScope.launch {
-		try {
-			block()
-		} finally {
-			try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-
-suspend inline infix operator fun <T> AsyncAioSocket.invoke(
-	@Suppress("REDUNDANT_INLINE_SUSPEND_FUNCTION_TYPE") block: suspend AsyncAioSocket.() -> T
-): T {
-	var exception: Throwable? = null
-	try {
-		return block()
-	} catch (e: Throwable) {
-		exception = e
-		throw e
-	} finally {
-		when (exception) {
-			null -> close()
-			else -> try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-}
-
-fun main() {
-	val port = 12345
-	val server = AsyncSocketServer(port) {
-		val buffer = ByteBuffer.allocate(1024)
-		while (true) {
-			buffer.clear()
-			read(buffer)
-			buffer.flip()
-			println("recv [${buffer.limit()}]")
-			write(buffer)
-		}
-	}
-	server.run()
-
-	val input = System.`in`.bufferedReader()
-	runBlocking {
-		val client = AsyncClient.connect("127.0.0.1", port)
-		val buffer = ByteBuffer.allocate(1024)
-		while (true) {
-			@Suppress("BlockingMethodInNonBlockingContext") val line = input.readLine()
-			println("sending [${line.length}]")
-			client.send(line)
-			while (client.read(buffer) == buffer.limit()) {
-				println("client recv [${buffer.position()}]")
-				println(String(buffer.array(), 0, buffer.position()))
-				buffer.clear()
-			}
-			println("client recv [${buffer.position()}]")
-			println(String(buffer.array(), 0, buffer.position()))
-			buffer.clear()
-		}
-	}
-}

socket/src/main/kotlin/cn/tursom/socket/AsyncCachedAioSocket.kt

@@ -1,196 +0,0 @@
-package cn.tursom.socket
-
-import cn.tursom.socket.AsyncAioSocket.Companion.defaultTimeout
-import cn.tursom.core.buf
-import cn.tursom.core.bytebuffer.NioAdvanceByteBuffer
-import cn.tursom.core.count
-import cn.tursom.core.put
-import cn.tursom.core.unSerialize
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
-import java.io.ByteArrayOutputStream
-import java.io.OutputStream
-import java.net.SocketTimeoutException
-import java.nio.ByteBuffer
-import java.nio.channels.AsynchronousSocketChannel
-import java.nio.channels.InterruptedByTimeoutException
-
-class AsyncCachedAioSocket(socketChannel: AsynchronousSocketChannel, readBuffer: ByteBuffer, val writeBuffer: ByteBuffer) : AsyncAioSocket(socketChannel) {
-	val readBuffer = NioAdvanceByteBuffer(readBuffer)
-	
-	constructor(socketChannel: AsynchronousSocketChannel) : this(socketChannel, ByteBuffer.allocate(1024), ByteBuffer.allocate(8))
-	
-	suspend fun write(timeout: Long = 0L): Int {
-		return write(writeBuffer, timeout)
-	}
-	
-	suspend fun read(timeout: Long = 0L): Int {
-		return read(readBuffer.buffer, timeout)
-	}
-}
-
-
-suspend inline fun AsyncCachedAioSocket.recv(
-	outputStream: OutputStream,
-	readTimeout: Long = 100L,
-	firstTimeout: Long = defaultTimeout
-) {
-	readBuffer.reset(outputStream)
-	
-	try {
-		read(firstTimeout)
-		readBuffer.reset(outputStream)
-		
-		while (read(readTimeout) > 0) {
-			readBuffer.reset(outputStream)
-		}
-	} catch (e: SocketTimeoutException) {
-	} catch (e: InterruptedByTimeoutException) {
-	}
-}
-
-suspend inline fun AsyncCachedAioSocket.recv(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = defaultTimeout
-): ByteArray {
-	val byteStream = ByteArrayOutputStream()
-	recv(byteStream, readTimeout, firstTimeout)
-	return byteStream.toByteArray()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvStr(
-	charset: String = "utf-8",
-	readTimeout: Long = 100L,
-	firstTimeout: Long = defaultTimeout
-): String {
-	val byteStream = ByteArrayOutputStream()
-	recv(byteStream, readTimeout, firstTimeout)
-	return byteStream.toString(charset)
-}
-
-suspend inline fun AsyncCachedAioSocket.recvChar(
-	readTimeout: Long = 100L
-): Char {
-	readBuffer.requireAvailableSize(2)
-	while (readBuffer.readableSize < 4) read(readTimeout)
-	return readBuffer.getChar()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvShort(
-	readTimeout: Long = 100L
-): Short {
-	readBuffer.requireAvailableSize(2)
-	while (readBuffer.readableSize < 8) read(readTimeout)
-	return readBuffer.getShort()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvInt(
-	readTimeout: Long = 100L
-): Int {
-	readBuffer.requireAvailableSize(4)
-	while (readBuffer.readableSize < 4) read(readTimeout)
-	return readBuffer.getInt()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvLong(
-	readTimeout: Long = 100L
-): Long {
-	readBuffer.requireAvailableSize(8)
-	while (readBuffer.readableSize < 8) read(readTimeout)
-	return readBuffer.getLong()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvFloat(
-	readTimeout: Long = 100L
-): Float {
-	readBuffer.requireAvailableSize(4)
-	while (readBuffer.readableSize < 4) read(readTimeout)
-	return readBuffer.getFloat()
-}
-
-suspend inline fun AsyncCachedAioSocket.recvDouble(
-	readTimeout: Long = 100L
-): Double {
-	readBuffer.requireAvailableSize(8)
-	while (readBuffer.readableSize < 8) read(readTimeout)
-	return readBuffer.getDouble()
-}
-
-@Suppress("UNCHECKED_CAST")
-suspend inline fun <T> AsyncCachedAioSocket.unSerializeObject(
-	readTimeout: Long = 100L,
-	firstTimeout: Long = defaultTimeout
-): T? {
-	val byteArrayOutputStream = ByteArrayOutputStream()
-	recv(byteArrayOutputStream, readTimeout, firstTimeout)
-	return unSerialize(byteArrayOutputStream.buf, 0, byteArrayOutputStream.count) as T?
-}
-
-suspend inline fun AsyncCachedAioSocket.send(message: Int) {
-	writeBuffer.clear()
-	writeBuffer.array().put(message, writeBuffer.arrayOffset())
-	writeBuffer.limit(4)
-	write()
-}
-
-suspend inline fun AsyncCachedAioSocket.send(message: Long) {
-	writeBuffer.clear()
-	writeBuffer.array().put(message, writeBuffer.arrayOffset())
-	writeBuffer.limit(8)
-	write()
-}
-
-inline fun <T> AsyncCachedAioSocket.use(crossinline block: suspend AsyncCachedAioSocket.() -> T): T {
-	var exception: Throwable? = null
-	try {
-		return runBlocking { block() }
-	} catch (e: Throwable) {
-		exception = e
-		throw e
-	} finally {
-		when (exception) {
-			null -> close()
-			else -> try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-}
-
-inline infix fun AsyncCachedAioSocket.useCachedNonBlock(crossinline block: suspend AsyncCachedAioSocket.() -> Unit) =
-	GlobalScope.launch {
-		try {
-			block()
-		} finally {
-			try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-
-
-suspend inline infix operator fun <T> AsyncCachedAioSocket.invoke(
-	@Suppress("REDUNDANT_INLINE_SUSPEND_FUNCTION_TYPE") block: suspend AsyncCachedAioSocket.() -> T
-): T {
-	var exception: Throwable? = null
-	try {
-		return block()
-	} catch (e: Throwable) {
-		exception = e
-		throw e
-	} finally {
-		when (exception) {
-			null -> close()
-			else -> try {
-				close()
-			} catch (closeException: Throwable) {
-				// cause.addSuppressed(closeException) // ignored here
-			}
-		}
-	}
-}

socket/src/main/kotlin/cn/tursom/socket/AsyncNioSocket.kt

@@ -1,239 +0,0 @@
-package cn.tursom.socket
-
-import cn.tursom.socket.niothread.INioThread
-import cn.tursom.core.timer.TimerTask
-import cn.tursom.core.timer.WheelTimer
-import java.nio.ByteBuffer
-import java.nio.channels.SelectionKey
-import java.nio.channels.SocketChannel
-import java.util.concurrent.TimeoutException
-import kotlin.coroutines.Continuation
-import kotlin.coroutines.resume
-import kotlin.coroutines.resumeWithException
-import kotlin.coroutines.suspendCoroutine
-
-/**
- * 利用 SelectionKey 的 attachment 进行状态的传输
- * 导致该类无法利用 SelectionKey 的 attachment
- * 但是对于一般的应用而言是足够使用的
- */
-class AsyncNioSocket(override val key: SelectionKey, override val nioThread: INioThread) : IAsyncNioSocket {
-	override val channel: SocketChannel = key.channel() as SocketChannel
-
-	override suspend fun read(buffer: ByteBuffer): Int {
-		if (buffer.remaining() == 0) return -1
-		return try {
-			suspendCoroutine {
-				key.attach(SingleContext(buffer, it))
-				readMode()
-				nioThread.wakeup()
-			}
-		} catch (e: Exception) {
-			waitMode()
-			throw RuntimeException(e)
-		}
-	}
-
-	override suspend fun read(buffer: Array<out ByteBuffer>): Long {
-		if (buffer.size == 0) return -1
-		return try {
-			suspendCoroutine {
-				key.attach(MultiContext(buffer, it))
-				readMode()
-				nioThread.wakeup()
-			}
-		} catch (e: Exception) {
-			waitMode()
-			throw RuntimeException(e)
-		}
-	}
-
-	override suspend fun write(buffer: ByteBuffer): Int {
-		if (buffer.remaining() == 0) return -1
-		return try {
-			suspendCoroutine {
-				key.attach(SingleContext(buffer, it))
-				writeMode()
-				nioThread.wakeup()
-			}
-		} catch (e: Exception) {
-			waitMode()
-			throw Exception(e)
-		}
-	}
-
-	override suspend fun write(buffer: Array<out ByteBuffer>): Long {
-		if (buffer.isEmpty()) return -1
-		return try {
-			suspendCoroutine {
-				key.attach(MultiContext(buffer, it))
-				writeMode()
-				nioThread.wakeup()
-			}
-		} catch (e: Exception) {
-			waitMode()
-			throw Exception(e)
-		}
-	}
-
-	override suspend fun read(buffer: ByteBuffer, timeout: Long): Int {
-		if (timeout <= 0) return read(buffer)
-		if (buffer.remaining() == 0) return -1
-		return try {
-			val result: Int = suspendCoroutine {
-				key.attach(SingleContext(buffer, it, timer.exec(timeout) {
-					try {
-						it.resumeWithException(TimeoutException())
-					} catch (e: Exception) {
-					}
-				}))
-				readMode()
-				nioThread.wakeup()
-			}
-			result
-		} catch (e: Exception) {
-			waitMode()
-			throw RuntimeException(e)
-		}
-	}
-
-	override suspend fun read(buffer: Array<out ByteBuffer>, timeout: Long): Long {
-		if (timeout <= 0) return read(buffer)
-		if (buffer.isEmpty()) return -1
-		return try {
-			val result: Long = suspendCoroutine {
-				key.attach(MultiContext(buffer, it, timer.exec(timeout) {
-					try {
-						it.resumeWithException(TimeoutException())
-					} catch (e: Exception) {
-					}
-				}))
-				readMode()
-				nioThread.wakeup()
-			}
-			result
-		} catch (e: Exception) {
-			waitMode()
-			throw Exception(e)
-		}
-	}
-
-	override suspend fun write(buffer: ByteBuffer, timeout: Long): Int {
-		if (timeout <= 0) return write(buffer)
-		if (buffer.remaining() == 0) return -1
-		return try {
-			val result: Int = suspendCoroutine {
-				key.attach(SingleContext(buffer, it, timer.exec(timeout) {
-					try {
-						it.resumeWithException(TimeoutException())
-					} catch (e: Exception) {
-					}
-				}))
-				writeMode()
-				nioThread.wakeup()
-			}
-			result
-		} catch (e: Exception) {
-			waitMode()
-			throw Exception(e)
-		}
-	}
-
-	override suspend fun write(buffer: Array<out ByteBuffer>, timeout: Long): Long {
-		if (timeout <= 0) return write(buffer)
-		if (buffer.isEmpty()) return -1
-		return try {
-			val result: Long = suspendCoroutine {
-				key.attach(MultiContext(buffer, it, timer.exec(timeout) {
-					try {
-						it.resumeWithException(TimeoutException())
-					} catch (e: Exception) {
-					}
-				}))
-				writeMode()
-				nioThread.wakeup()
-			}
-			result
-		} catch (e: Exception) {
-			waitMode()
-			throw Exception(e)
-		}
-	}
-
-	override fun close() {
-		nioThread.execute {
-			channel.close()
-			key.cancel()
-		}
-	}
-
-	interface Context {
-		val cont: Continuation<*>
-		val timeoutTask: TimerTask? get() = null
-	}
-
-	class SingleContext(
-		val buffer: ByteBuffer,
-		override val cont: Continuation<Int>,
-		override val timeoutTask: TimerTask? = null
-	) : Context
-
-	class MultiContext(
-		val buffer: Array<out ByteBuffer>,
-		override val cont: Continuation<Long>,
-		override val timeoutTask: TimerTask? = null
-	) : Context
-
-	companion object {
-		val nioSocketProtocol = object : INioProtocol {
-			override fun handleConnect(key: SelectionKey, nioThread: INioThread) {}
-
-			override fun handleRead(key: SelectionKey, nioThread: INioThread) {
-				key.interestOps(0)
-				val context = key.attachment() as Context
-				context.timeoutTask?.cancel()
-				if (context is SingleContext) {
-					val channel = key.channel() as SocketChannel
-					val readSize = channel.read(context.buffer)
-					context.cont.resume(readSize)
-				} else {
-					context as MultiContext
-					val channel = key.channel() as SocketChannel
-					val readSize = channel.read(context.buffer)
-					context.cont.resume(readSize)
-				}
-			}
-
-			override fun handleWrite(key: SelectionKey, nioThread: INioThread) {
-				key.interestOps(0)
-				val context = key.attachment() as Context
-				context.timeoutTask?.cancel()
-				if (context is SingleContext) {
-					val channel = key.channel() as SocketChannel
-					val readSize = channel.write(context.buffer)
-					context.cont.resume(readSize)
-				} else {
-					context as MultiContext
-					val channel = key.channel() as SocketChannel
-					val readSize = channel.write(context.buffer)
-					context.cont.resume(readSize)
-				}
-			}
-
-			override fun exceptionCause(key: SelectionKey, nioThread: INioThread, e: Throwable) {
-				key.interestOps(0)
-				val context = key.attachment() as Context?
-				if (context != null)
-					context.cont.resumeWithException(e)
-				else {
-					key.cancel()
-					key.channel().close()
-					e.printStackTrace()
-				}
-			}
-		}
-
-		//val timer = StaticWheelTimer.timer
-		val timer = WheelTimer.timer
-	}
-}

socket/src/main/kotlin/cn/tursom/socket/BaseSocket.kt

@@ -10,202 +10,68 @@ import java.net.SocketTimeoutException
  */
 @Suppress("unused", "MemberVisibilityCanBePrivate")
 open class BaseSocket(
-	private val socket: Socket,
-	private val timeout: Int = Companion.timeout
+    val socket: Socket,
+    val timeout: Int = Companion.timeout
 ) : Closeable {
-	
-	val address = socket.inetAddress?.toString()?.drop(1) ?: "0.0.0.0"
-	val port = socket.port
-	val localPort = socket.localPort
-	private val inputStream = socket.getInputStream()!!
-	private val outputStream = socket.getOutputStream()!!
-	
-	fun send(message: String?) {
-		send((message ?: return).toByteArray())
-	}
-	
-	fun send(message: ByteArray?) {
-		outputStream.write(message ?: return)
-	}
-	
-	fun send(message: Int) {
-		val buffer = ByteArray(4)
-		buffer.put(message)
-		send(buffer)
-	}
-	
-	fun send(message: Long) {
-		val buffer = ByteArray(8)
-		buffer.put(message)
-		send(buffer)
-	}
-	
-	fun sendObject(obj: Any?): Boolean {
-		send(serialize(obj ?: return false) ?: return false)
-		return true
-	}
-	
-	inline fun <reified T> recvObject(): T? {
-		return try {
-			unSerialize(recv()) as T
-		} catch (e: Exception) {
-			null
-		}
-	}
-	
-	fun recvString(
-		readTimeout: Int = 100,
-		firstTimeout: Int = timeout
-	): String {
-		return recv(readTimeout, firstTimeout).toUTF8String()
-	}
-	
-	fun recvString(
-		maxsize: Int,
-		readTimeout: Int = 100,
-		firstTimeout: Int = timeout
-	): String {
-		return recv(maxsize, readTimeout, firstTimeout).toUTF8String()
-	}
-	
-	fun recvInt(
-		timeout1: Int = timeout
-	): Int? {
-		val buffer = ByteArray(4)
-		socket.soTimeout = timeout1
-		var sTime = System.currentTimeMillis()
-		//读取数据
-		var rSize = inputStream.read(buffer, 0, 4)
-		while (rSize < 4) {
-			val sTime2 = System.currentTimeMillis()
-			socket.soTimeout -= (sTime2 - sTime).toInt()
-			sTime = sTime2
-			val sReadSize = inputStream.read(buffer, rSize, 8 - rSize)
-			if (sReadSize <= 0) {
-				break
-			} else {
-				rSize += sReadSize
-			}
-		}
-		return buffer.toInt()
-	}
-	
-	fun recvLong(
-		timeout1: Int = timeout
-	): Long? {
-		val buffer = ByteArray(8)
-		socket.soTimeout = timeout1
-		var sTime = System.currentTimeMillis()
-		//读取数据
-		var rSize = inputStream.read(buffer, 0, 8)
-		while (rSize < 4) {
-			val sTime2 = System.currentTimeMillis()
-			socket.soTimeout -= (sTime2 - sTime).toInt()
-			sTime = sTime2
-			val sReadSize = inputStream.read(buffer, rSize, 8 - rSize)
-			if (sReadSize <= 0) {
-				break
-			} else {
-				rSize += sReadSize
-			}
-		}
-		return buffer.toLong()
-	}
-	
-	fun recv(
-		readTimeout: Int = 100,
-		firstTimeout: Int = timeout
-	): ByteArray {
-		val outputStream = ByteArrayOutputStream()
-		recv(outputStream, readTimeout, firstTimeout)
-		return outputStream.toByteArray()
-	}
-	
-	fun recv(
-		maxsize: Int,
-		readTimeout: Int = 100,
-		firstTimeout: Int = timeout
-	): ByteArray {
-		val buffer = ByteArray(maxsize)
-		var readSize = 0
-		socket.soTimeout = firstTimeout
-		
-		try {
-			readSize = inputStream.read(buffer)
-			
-			socket.soTimeout = readTimeout
-			while (readSize < buffer.size) {
-				val sReadSize = inputStream.read(buffer, readSize, buffer.size - readSize)
-				if (sReadSize <= 0) {
-					break
-				} else {
-					readSize += sReadSize
-				}
-			}
-		} catch (e: SocketTimeoutException) {
-		}
-		return buffer.copyOf(readSize)
-	}
-	
-	fun recv(
-		outputStream: OutputStream,
-		readTimeout: Int = 100,
-		firstTimeout: Int = timeout
-	) {
-		val buffer = ByteArray(1024)
-		socket.soTimeout = firstTimeout
-		
-		try {
-			val readSize = inputStream.read(buffer)
-			if (readSize < 0) {
-				throw IOException("cannot read data")
-			}
-			outputStream.write(buffer, 0, readSize)
-			socket.soTimeout = readTimeout
-			while (true) {
-				val sReadSize = inputStream.read(buffer)
-				if (sReadSize <= 0) {
-					break
-				} else {
-					outputStream.write(buffer, 0, readSize)
-				}
-			}
-		} catch (e: SocketTimeoutException) {
-		}
-	}
-	
-	override fun close() {
-		closeSocket()
-	}
-	
-	protected fun closeSocket() {
-		if (!socket.isClosed) {
-			closeInputStream()
-			closeOutputStream()
-			socket.close()
-		}
-	}
-	
-	private fun closeInputStream() {
-		try {
-			inputStream.close()
-		} catch (e: Exception) {
-		}
-	}
-	
-	private fun closeOutputStream() {
-		try {
-			outputStream.close()
-		} catch (e: Exception) {
-		}
-	}
-	
-	fun isConnected(): Boolean {
-		return socket.isConnected
-	}
-	
-	companion object Companion {
-		const val defaultReadSize: Int = 1024 * 8
-		const val timeout: Int = 60 * 1000
-	}
+
+    val address = socket.inetAddress?.toString()?.drop(1) ?: "0.0.0.0"
+    val port = socket.port
+    val localPort = socket.localPort
+    val inputStream = socket.getInputStream()!!
+    val outputStream = socket.getOutputStream()!!
+
+    fun send(message: String?) {
+        send((message ?: return).toByteArray())
+    }
+
+    fun send(message: ByteArray?) {
+        outputStream.write(message ?: return)
+    }
+
+    fun send(message: Int) {
+        val buffer = ByteArray(4)
+        buffer.put(message)
+        send(buffer)
+    }
+
+    fun send(message: Long) {
+        val buffer = ByteArray(8)
+        buffer.put(message)
+        send(buffer)
+    }
+
+    override fun close() {
+        closeSocket()
+    }
+
+    protected fun closeSocket() {
+        if (!socket.isClosed) {
+            closeInputStream()
+            closeOutputStream()
+            socket.close()
+        }
+    }
+
+    private fun closeInputStream() {
+        try {
+            inputStream.close()
+        } catch (e: Exception) {
+        }
+    }
+
+    private fun closeOutputStream() {
+        try {
+            outputStream.close()
+        } catch (e: Exception) {
+        }
+    }
+
+    fun isConnected(): Boolean {
+        return socket.isConnected
+    }
+
+    companion object Companion {
+        const val defaultReadSize: Int = 1024 * 8
+        const val timeout: Int = 60 * 1000
+    }
 }

socket/src/main/kotlin/cn/tursom/socket/SocketClient.kt

@@ -1,6 +1,5 @@
-package cn.tursom.socket.client
+package cn.tursom.socket
 
-import cn.tursom.socket.BaseSocket
 import java.io.IOException
 import java.net.Socket
 import java.net.SocketException

socket/src/main/kotlin/cn/tursom/socket/server/GroupNioServer.kt

@@ -1,11 +1,10 @@
-package cn.tursom.socket.server.nio
+package cn.tursom.socket.server
 
 import cn.tursom.socket.INioProtocol
 import cn.tursom.socket.niothread.INioThread
 import cn.tursom.socket.niothread.IWorkerGroup
 import cn.tursom.socket.niothread.ThreadPoolNioThread
 import cn.tursom.socket.niothread.ThreadPoolWorkerGroup
-import cn.tursom.socket.server.ISocketServer
 import java.net.InetSocketAddress
 import java.nio.channels.SelectionKey
 import java.nio.channels.Selector
@@ -48,33 +47,33 @@ class GroupNioServer(
 		val nioThread = ThreadPoolNioThread("nioAccepter") { nioThread ->
 			val selector = nioThread.selector
 			if (selector.isOpen) {
-				forEachKey(selector) { key ->
-					try {
-						when {
-							key.isAcceptable -> {
-								val serverChannel = key.channel() as ServerSocketChannel
-								var channel = serverChannel.accept()
-								while (channel != null) {
-									channel.configureBlocking(false)
-									workerGroup.register(channel) { (key, thread) ->
-										protocol.handleConnect(key, thread)
-									}
-									channel = serverChannel.accept()
-								}
-							}
-						}
-					} catch (e: Throwable) {
-						try {
-							protocol.exceptionCause(key, nioThread, e)
-						} catch (e1: Throwable) {
-							e.printStackTrace()
-							e1.printStackTrace()
-							key.cancel()
-							key.channel().close()
-						}
-					}
-					nioThread.execute(this)
-				}
+                forEachKey(selector) { key ->
+                    try {
+                        when {
+                            key.isAcceptable -> {
+                                val serverChannel = key.channel() as ServerSocketChannel
+                                var channel = serverChannel.accept()
+                                while (channel != null) {
+                                    channel.configureBlocking(false)
+                                    workerGroup.register(channel) { (key, thread) ->
+                                        protocol.handleConnect(key, thread)
+                                    }
+                                    channel = serverChannel.accept()
+                                }
+                            }
+                        }
+                    } catch (e: Throwable) {
+                        try {
+                            protocol.exceptionCause(key, nioThread, e)
+                        } catch (e1: Throwable) {
+                            e.printStackTrace()
+                            e1.printStackTrace()
+                            key.cancel()
+                            key.channel().close()
+                        }
+                    }
+                    nioThread.execute(this)
+                }
 			}
 		}
 		listenThreads.add(nioThread)

socket/src/main/kotlin/cn/tursom/socket/server/MultithreadingSocketServer.kt

@@ -4,44 +4,49 @@ import cn.tursom.socket.BaseSocket
 import java.net.ServerSocket
 
 class MultithreadingSocketServer(
-	private val serverSocket: ServerSocket,
-	private val threadNumber: Int = cpuNumber,
-	val exception: Exception.() -> Unit = {
-		printStackTrace()
-	},
-	handler: BaseSocket.() -> Unit
+    private val serverSocket: ServerSocket,
+    private val threadNumber: Int = cpuNumber,
+    val exception: Exception.() -> Unit = {
+        printStackTrace()
+    },
+    handler: BaseSocket.() -> Unit
 ) : SocketServer(handler) {
-	
-	constructor(
-		port: Int,
-		threadNumber: Int = cpuNumber,
-		exception: Exception.() -> Unit = {
-			printStackTrace()
-		},
-		handler: BaseSocket.() -> Unit
-	) : this(ServerSocket(port), threadNumber, exception, handler)
-	
-	private val threadList = ArrayList<Thread>()
-	
-	override fun run() {
-		for (i in 1..threadNumber) {
-			val thread = Thread {
-				while (true) {
-					serverSocket.accept().use {
-						try {
-							BaseSocket(it).handler()
-						} catch (e: Exception) {
-							e.exception()
-						}
-					}
-				}
-			}
-			thread.start()
-			threadList.add(thread)
-		}
-	}
-	
-	override fun close() {
-		serverSocket.close()
-	}
+
+    constructor(
+        port: Int,
+        threadNumber: Int = cpuNumber,
+        exception: Exception.() -> Unit = {
+            printStackTrace()
+        },
+        handler: BaseSocket.() -> Unit
+    ) : this(ServerSocket(port), threadNumber, exception, handler)
+
+    constructor(
+        port: Int,
+        handler: BaseSocket.() -> Unit
+    ) : this(port, cpuNumber, { printStackTrace() }, handler)
+
+    private val threadList = ArrayList<Thread>()
+
+    override fun run() {
+        for (i in 1..threadNumber) {
+            val thread = Thread {
+                while (true) {
+                    serverSocket.accept().use {
+                        try {
+                            BaseSocket(it).handler()
+                        } catch (e: Exception) {
+                            e.exception()
+                        }
+                    }
+                }
+            }
+            thread.start()
+            threadList.add(thread)
+        }
+    }
+
+    override fun close() {
+        serverSocket.close()
+    }
 }

socket/src/main/kotlin/cn/tursom/socket/server/NioServer.kt

@@ -0,0 +1,97 @@
+package cn.tursom.socket.server
+
+import cn.tursom.socket.INioProtocol
+import cn.tursom.socket.niothread.INioThread
+import cn.tursom.socket.niothread.WorkerLoopNioThread
+import java.net.InetSocketAddress
+import java.nio.channels.SelectionKey
+import java.nio.channels.ServerSocketChannel
+import java.util.concurrent.ConcurrentLinkedDeque
+
+/**
+ * 工作在单线程上的 Nio 服务器。
+ */
+class NioServer(
+    val port: Int,
+    private val protocol: INioProtocol,
+    backLog: Int = 50,
+    val nioThreadGenerator: (threadName: String, workLoop: (thread: INioThread) -> Unit) -> INioThread
+) : ISocketServer {
+    private val listenChannel = ServerSocketChannel.open()
+    private val threadList = ConcurrentLinkedDeque<INioThread>()
+
+    init {
+        listenChannel.socket().bind(InetSocketAddress(port), backLog)
+        listenChannel.configureBlocking(false)
+    }
+
+    constructor(
+        port: Int,
+        protocol: INioProtocol,
+        backLog: Int = 50
+    ) : this(port, protocol, backLog, { name, workLoop ->
+        WorkerLoopNioThread(name, workLoop = workLoop)
+    })
+
+    override fun run() {
+        val nioThread = nioThreadGenerator("nio worker", LoopHandler(protocol)::handle)
+        nioThread.register(listenChannel, SelectionKey.OP_ACCEPT) {}
+        threadList.add(nioThread)
+    }
+
+    override fun close() {
+        listenChannel.close()
+        threadList.forEach {
+            it.close()
+        }
+    }
+
+    class LoopHandler(val protocol: INioProtocol) {
+        fun handle(nioThread: INioThread) {
+            val selector = nioThread.selector
+            if (selector.isOpen) {
+                if (selector.select(TIMEOUT) != 0) {
+                    val keyIter = selector.selectedKeys().iterator()
+                    while (keyIter.hasNext()) run whileBlock@{
+                        val key = keyIter.next()
+                        keyIter.remove()
+                        try {
+                            when {
+                                key.isAcceptable -> {
+                                    val serverChannel = key.channel() as ServerSocketChannel
+                                    var channel = serverChannel.accept()
+                                    while (channel != null) {
+                                        channel.configureBlocking(false)
+                                        nioThread.register(channel, 0) {
+                                            protocol.handleConnect(it, nioThread)
+                                        }
+                                        channel = serverChannel.accept()
+                                    }
+                                }
+                                key.isReadable -> {
+                                    protocol.handleRead(key, nioThread)
+                                }
+                                key.isWritable -> {
+                                    protocol.handleWrite(key, nioThread)
+                                }
+                            }
+                        } catch (e: Throwable) {
+                            try {
+                                protocol.exceptionCause(key, nioThread, e)
+                            } catch (e1: Throwable) {
+                                e.printStackTrace()
+                                e1.printStackTrace()
+                                key.cancel()
+                                key.channel().close()
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    companion object {
+        private const val TIMEOUT = 1000L
+    }
+}

socket/src/main/kotlin/cn/tursom/socket/server/ServerHandler.kt

@@ -1,64 +0,0 @@
-package cn.tursom.socket.server
-
-import cn.tursom.socket.BaseSocket
-import java.net.Socket
-
-/**
- * ServerHandler请求处理类
- * 通过重载handle()函数处理逻辑
- * recvString()提供了网络通讯常见的recv函数，避免BuffedReader.getLine造成的阻塞
- * 自动关闭套接字，自动处理异常（全局）
- * 通拥有较好的异常处理体系，可通过异常实现基本的逻辑
- * 可以处理异常的同时给客户端发送异常信息，通过重载ServerException.code的getter实现
- */
- class ServerHandler(
-	socket: Socket,
-	val serverError: ByteArray=Companion.serverError,
-	timeout: Int = BaseSocket.timeout,
-	val handler:BaseSocket.()->Unit
-) : Runnable, BaseSocket(socket, timeout) {
-	init {
-		if (socket.isClosed) {
-			throw SocketClosedException()
-		}
-	}
-	
-	 override fun run() {
-		try {
-			handler()
-		} catch (e: ServerException) {
-			if (e.message == null)
-				e.printStackTrace()
-			else
-				System.err.println("$address: ${e::class.java}: ${e.message}")
-			
-			try {
-				send(serverError)
-			} catch (e: SocketClosedException) {
-				System.err.println("$address: ${e::class.java}: ${e.message}")
-			}
-			
-		} catch (e: Exception) {
-			e.printStackTrace()
-			try {
-				send(serverError)
-			} catch (e: SocketClosedException) {
-				System.err.println("$address: ${e::class.java}: ${e.message}")
-			}
-		}
-		closeSocket()
-		println("$address: connection closed")
-	}
-	
-	open class ServerException(s: String? = null) : Exception(s) {
-		open val code: ByteArray?
-			get() = null
-	}
-	
-	class SocketClosedException(s: String? = null) : ServerException(s)
-	
-	companion object Companion {
-		val serverError = "server error".toByteArray()
-	}
-}
-

socket/src/main/kotlin/cn/tursom/socket/server/SingleThreadSocketServer.kt

@@ -5,42 +5,47 @@ import java.net.ServerSocket
 import java.net.SocketException
 
 class SingleThreadSocketServer(
-	private val serverSocket: ServerSocket,
-	val exception: Exception.() -> Unit = { printStackTrace() },
-	handler: BaseSocket.() -> Unit
+    private val serverSocket: ServerSocket,
+    val exception: Exception.() -> Unit = { printStackTrace() },
+    handler: BaseSocket.() -> Unit
 ) : SocketServer(handler) {
-	
-	constructor(
-		port: Int,
-		exception: Exception.() -> Unit = { printStackTrace() },
-		handler: BaseSocket.() -> Unit
-	) : this(ServerSocket(port), exception, handler)
-	
-	override fun run() {
-		while (!serverSocket.isClosed) {
-			try {
-				serverSocket.accept().use {
-					try {
-						BaseSocket(it).handler()
-					} catch (e: Exception) {
-						e.exception()
-					}
-				}
-			} catch (e: SocketException) {
-				if (e.message == "Socket closed" || e.message == "cn.tursom.socket closed") {
-					break
-				} else {
-					e.exception()
-				}
-			}
-		}
-	}
-	
-	override fun close() {
-		try {
-			serverSocket.close()
-		} catch (e: Exception) {
-			e.printStackTrace()
-		}
-	}
+
+    constructor(
+        port: Int,
+        exception: Exception.() -> Unit = { printStackTrace() },
+        handler: BaseSocket.() -> Unit
+    ) : this(ServerSocket(port), exception, handler)
+
+    constructor(
+        port: Int,
+        handler: BaseSocket.() -> Unit
+    ) : this(port, { printStackTrace() }, handler)
+
+    override fun run() {
+        while (!serverSocket.isClosed) {
+            try {
+                serverSocket.accept().use {
+                    try {
+                        BaseSocket(it).handler()
+                    } catch (e: Exception) {
+                        e.exception()
+                    }
+                }
+            } catch (e: SocketException) {
+                if (e.message == "Socket closed" || e.message == "cn.tursom.socket closed") {
+                    break
+                } else {
+                    e.exception()
+                }
+            }
+        }
+    }
+
+    override fun close() {
+        try {
+            serverSocket.close()
+        } catch (e: Exception) {
+            e.printStackTrace()
+        }
+    }
 }

socket/src/main/kotlin/cn/tursom/socket/server/ThreadPoolSocketServer.kt

@@ -2,7 +2,6 @@ package cn.tursom.socket.server
 
 import cn.tursom.core.getTAG
 import cn.tursom.socket.BaseSocket
-import java.io.File
 import java.io.IOException
 import java.net.ServerSocket
 import java.net.Socket
@@ -47,121 +46,118 @@ open class ThreadPoolSocketServer
 	queueSize: Int = 1,
 	keepAliveTime: Long = 60_000L,
 	timeUnit: TimeUnit = TimeUnit.MILLISECONDS,
-	startImmediately: Boolean = false,
 	handler: BaseSocket.() -> Unit
 ) : SocketServer(handler) {
-	
-	var socket = Socket()
-	private val pool: ThreadPoolExecutor
-	private var serverSocket: ServerSocket
 
-	/**
-	 * 为了在构造函数中自动启动服务，我们需要封闭start()，防止用户重载start()
-	 */
-	private fun start() {
-		Thread(this).start()
-	}
-	
-	/**
-	 * 主要作用：
-	 * 循环接受连接请求
-	 * 讲接收的连接交给handler处理
-	 * 连接初期异常处理
-	 * 自动关闭套接字服务器与线程池
-	 */
-	final override fun run() {
-		while (!serverSocket.isClosed) {
-			try {
-				socket = serverSocket.accept()
-				println("$TAG: run(): get connect: $socket")
-				pool.execute {
-					socket.use {
-						BaseSocket(it).handler()
-					}
-				}
-			} catch (e: IOException) {
-				if (pool.isShutdown || serverSocket.isClosed) {
-					System.err.println("server closed")
-					break
-				}
-				e.printStackTrace()
-			} catch (e: SocketException) {
-				e.printStackTrace()
-				break
-			} catch (e: RejectedExecutionException) {
-				socket.getOutputStream()?.write(poolIsFull)
-			} catch (e: Exception) {
-				e.printStackTrace()
-				break
-			}
-		}
-		whenClose()
-		close()
-		System.err.println("server closed")
-	}
-	
-	/**
-	 * 关闭服务器套接字
-	 */
-	private fun closeServer() {
-		if (!serverSocket.isClosed) {
-			serverSocket.close()
-		}
-	}
-	
-	/**
-	 * 关闭线程池
-	 */
-	private fun shutdownPool() {
-		if (!pool.isShutdown) {
-			pool.shutdown()
-		}
-	}
-	
-	/**
-	 * 服务器是否已经关闭
-	 */
-	@Suppress("unused")
-	fun isClosed() = pool.isShutdown || serverSocket.isClosed
-	
-	/**
-	 * 关闭服务器
-	 */
-	override fun close() {
-		shutdownPool()
-		closeServer()
-	}
-	
-	/**
-	 * 关闭服务器时执行
-	 */
-	open fun whenClose() {
-	}
-	
-	/**
-	 * 线程池满时返回给客户端的信息
-	 */
-	open val poolIsFull
-		get() = Companion.poolIsFull
+	constructor(
+		port: Int,
+		handler: BaseSocket.() -> Unit
+	) : this(port, 1, 1, 60_000L, TimeUnit.MILLISECONDS, handler)
 
-	private data class ServerConfigData(
-		val port: Int = 0,
-		val threads: Int = 1,
-		val queueSize: Int = 1,
-		val timeout: Long = 0L,
-		val startImmediately: Boolean = false
-	)
-	
-	companion object {
-		val TAG = getTAG(this::class.java)
-		val poolIsFull = "server pool is full".toByteArray()
-	}
+    var socket = Socket()
+    private val pool: ThreadPoolExecutor =
+        ThreadPoolExecutor(threads, threads, keepAliveTime, timeUnit, LinkedBlockingQueue(queueSize))
+    private var serverSocket: ServerSocket = ServerSocket(port)
 
-	init {
-		pool = ThreadPoolExecutor(threads, threads, keepAliveTime, timeUnit, LinkedBlockingQueue(queueSize))
-		serverSocket = ServerSocket(port)
-		if (startImmediately) {
-			start()
-		}
-	}
+    /**
+     * 为了在构造函数中自动启动服务，我们需要封闭start()，防止用户重载start()
+     */
+    private fun start() {
+        Thread(this).start()
+    }
+
+    /**
+     * 主要作用：
+     * 循环接受连接请求
+     * 讲接收的连接交给handler处理
+     * 连接初期异常处理
+     * 自动关闭套接字服务器与线程池
+     */
+    final override fun run() {
+        while (!serverSocket.isClosed) {
+            try {
+                socket = serverSocket.accept()
+                println("$TAG: run(): get connect: $socket")
+                pool.execute {
+                    socket.use {
+                        BaseSocket(it).handler()
+                    }
+                }
+            } catch (e: IOException) {
+                if (pool.isShutdown || serverSocket.isClosed) {
+                    System.err.println("server closed")
+                    break
+                }
+                e.printStackTrace()
+            } catch (e: SocketException) {
+                e.printStackTrace()
+                break
+            } catch (e: RejectedExecutionException) {
+                socket.getOutputStream()?.write(poolIsFull)
+            } catch (e: Exception) {
+                e.printStackTrace()
+                break
+            }
+        }
+        whenClose()
+        close()
+        System.err.println("server closed")
+    }
+
+    /**
+     * 关闭服务器套接字
+     */
+    private fun closeServer() {
+        if (!serverSocket.isClosed) {
+            serverSocket.close()
+        }
+    }
+
+    /**
+     * 关闭线程池
+     */
+    private fun shutdownPool() {
+        if (!pool.isShutdown) {
+            pool.shutdown()
+        }
+    }
+
+    /**
+     * 服务器是否已经关闭
+     */
+    @Suppress("unused")
+    fun isClosed() = pool.isShutdown || serverSocket.isClosed
+
+    /**
+     * 关闭服务器
+     */
+    override fun close() {
+        shutdownPool()
+        closeServer()
+    }
+
+    /**
+     * 关闭服务器时执行
+     */
+    open fun whenClose() {
+    }
+
+    /**
+     * 线程池满时返回给客户端的信息
+     */
+    open val poolIsFull
+        get() = Companion.poolIsFull
+
+    private data class ServerConfigData(
+        val port: Int = 0,
+        val threads: Int = 1,
+        val queueSize: Int = 1,
+        val timeout: Long = 0L,
+        val startImmediately: Boolean = false
+    )
+
+    companion object {
+        val TAG = getTAG(this::class.java)
+        val poolIsFull = "server pool is full".toByteArray()
+    }
 }

socket/src/main/kotlin/cn/tursom/socket/server/nio/NioServer.kt

@@ -1,92 +0,0 @@
-package cn.tursom.socket.server.nio
-
-import cn.tursom.socket.INioProtocol
-import cn.tursom.socket.niothread.INioThread
-import cn.tursom.socket.niothread.WorkerLoopNioThread
-import cn.tursom.socket.server.ISocketServer
-import java.net.InetSocketAddress
-import java.nio.channels.SelectionKey
-import java.nio.channels.ServerSocketChannel
-import java.util.concurrent.ConcurrentLinkedDeque
-
-/**
- * 工作在单线程上的 Nio 服务器。
- */
-class NioServer(
-	val port: Int,
-	private val protocol: INioProtocol,
-	backLog: Int = 50,
-	val nioThreadGenerator: (threadName: String, workLoop: (thread: INioThread) -> Unit) -> INioThread = { name, workLoop ->
-		WorkerLoopNioThread(name, workLoop = workLoop)
-	}
-) : ISocketServer {
-	private val listenChannel = ServerSocketChannel.open()
-	private val threadList = ConcurrentLinkedDeque<INioThread>()
-
-	init {
-		listenChannel.socket().bind(InetSocketAddress(port), backLog)
-		listenChannel.configureBlocking(false)
-	}
-
-	override fun run() {
-		val nioThread = nioThreadGenerator("nio worker", LoopHandler(protocol)::handle)
-		nioThread.register(listenChannel, SelectionKey.OP_ACCEPT) {}
-		threadList.add(nioThread)
-	}
-
-	override fun close() {
-		listenChannel.close()
-		threadList.forEach {
-			it.close()
-		}
-	}
-
-	class LoopHandler(val protocol: INioProtocol) {
-		fun handle(nioThread: INioThread) {
-			val selector = nioThread.selector
-			if (selector.isOpen) {
-				if (selector.select(TIMEOUT) != 0) {
-					val keyIter = selector.selectedKeys().iterator()
-					while (keyIter.hasNext()) run whileBlock@{
-						val key = keyIter.next()
-						keyIter.remove()
-						try {
-							when {
-								key.isAcceptable -> {
-									val serverChannel = key.channel() as ServerSocketChannel
-									var channel = serverChannel.accept()
-									while (channel != null) {
-										channel.configureBlocking(false)
-										nioThread.register(channel, 0) {
-											protocol.handleConnect(it, nioThread)
-										}
-										channel = serverChannel.accept()
-									}
-								}
-								key.isReadable -> {
-									protocol.handleRead(key, nioThread)
-								}
-								key.isWritable -> {
-									protocol.handleWrite(key, nioThread)
-								}
-							}
-						} catch (e: Throwable) {
-							try {
-								protocol.exceptionCause(key, nioThread, e)
-							} catch (e1: Throwable) {
-								e.printStackTrace()
-								e1.printStackTrace()
-								key.cancel()
-								key.channel().close()
-							}
-						}
-					}
-				}
-			}
-		}
-	}
-
-	companion object {
-		private const val TIMEOUT = 1000L
-	}
-}