add WebSocketFrameWrapper

build.gradle

@@ -1,5 +1,5 @@
 buildscript {
-    ext.kotlinVersion = '1.4.0'
+    ext.kotlinVersion = '1.4.21'
 
     repositories {
         mavenLocal()

utils/build.gradle

@@ -8,4 +8,5 @@ dependencies {
     // 计算对象大小
     api 'org.apache.lucene:lucene-core:4.0.0'
     api group: "io.netty", name: "netty-all", version: "4.1.43.Final"
+    api group: "io.netty", name: "netty-all", version: "4.1.43.Final"
 }

utils/src/main/kotlin/cn/tursom/utils/WebSocketFrameWrapper.kt

@@ -0,0 +1,22 @@
+package cn.tursom.utils
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
+import io.netty.channel.ChannelHandler
+import io.netty.channel.ChannelHandlerContext
+import io.netty.channel.ChannelOutboundHandlerAdapter
+import io.netty.channel.ChannelPromise
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
+
+@ChannelHandler.Sharable
+object WebSocketFrameWrapper : ChannelOutboundHandlerAdapter() {
+    override fun write(ctx: ChannelHandlerContext, msg: Any?, promise: ChannelPromise?) {
+        ctx.write(when (msg) {
+            is String -> TextWebSocketFrame(msg)
+            is ByteArray -> BinaryWebSocketFrame(Unpooled.wrappedBuffer(msg))
+            is ByteBuf -> BinaryWebSocketFrame(msg)
+            else -> msg
+        }, promise)
+    }
+}

utils/ws-client/src/main/kotlin/cn/tursom/ws/WebSocketClient.kt

@@ -1,6 +1,7 @@
 package cn.tursom.ws
 
 import cn.tursom.core.buffer.ByteBuffer
+import cn.tursom.utils.WebSocketFrameWrapper
 import cn.tursom.utils.bytebuffer.NettyByteBuffer
 import io.netty.bootstrap.Bootstrap
 import io.netty.buffer.ByteBuf
@@ -22,7 +23,7 @@ import io.netty.handler.ssl.util.InsecureTrustManagerFactory
 import java.net.URI
 
 
-class WebSocketClient(uri: String, val handler: WebSocketHandler) {
+class WebSocketClient(uri: String, val handler: WebSocketHandler, val autoWrap: Boolean = true) {
   private val uri: URI = URI.create(uri)
   internal var ch: Channel? = null
 
@@ -59,24 +60,27 @@ class WebSocketClient(uri: String, val handler: WebSocketHandler) {
         uri, WebSocketVersion.V13, null, false, DefaultHttpHeaders()
       ), this, handler
     )
-    val b = Bootstrap()
-    b.group(group)
+    val bootstrap = Bootstrap()
+    bootstrap.group(group)
       .channel(NioSocketChannel::class.java)
       .handler(object : ChannelInitializer<SocketChannel>() {
         override fun initChannel(ch: SocketChannel) {
-          val p = ch.pipeline()
+          val pipeline = ch.pipeline()
           if (sslCtx != null) {
-            p.addLast(sslCtx.newHandler(ch.alloc(), host, port))
+            pipeline.addLast(sslCtx.newHandler(ch.alloc(), host, port))
           }
-          p.addLast(
+          pipeline.addLast(
             HttpClientCodec(),
             HttpObjectAggregator(4096),
             WebSocketClientCompressionHandler.INSTANCE,
-            handler
+            handler,
           )
+          if (autoWrap) {
+            pipeline.addLast(WebSocketFrameWrapper)
+          }
         }
       })
-    b.connect(uri.host, port)
+    bootstrap.connect(uri.host, port)
     //handler.handshakeFuture().sync()
   }
 

web/netty-web/src/main/kotlin/cn/tursom/web/netty/NettyHttpServer.kt

@@ -1,5 +1,6 @@
 package cn.tursom.web.netty
 
+import cn.tursom.utils.WebSocketFrameWrapper
 import cn.tursom.web.HttpHandler
 import cn.tursom.web.HttpServer
 import cn.tursom.web.WebSocketHandler
@@ -31,6 +32,7 @@ class NettyHttpServer(
   var writeTimeout: Int? = null,
   decodeType: NettyHttpDecodeType = if (webSocketPath.iterator().hasNext()) NettyHttpDecodeType.FULL_HTTP else NettyHttpDecodeType.MULTI_PART,
   backlog: Int = 1024,
+  val wrapWebSocketFrame: Boolean = false,
 ) : HttpServer {
   constructor(
     port: Int,
@@ -40,13 +42,15 @@ class NettyHttpServer(
     readTimeout: Int? = 60,
     writeTimeout: Int? = null,
     decodeType: NettyHttpDecodeType = if (webSocketPath.iterator().hasNext()) NettyHttpDecodeType.FULL_HTTP else NettyHttpDecodeType.MULTI_PART,
+    backlog: Int = 1024,
+    wrapWebSocketFrame: Boolean = false,
     handler: (content: NettyHttpContent) -> Unit,
   ) : this(
     port,
     object : HttpHandler<NettyHttpContent, NettyExceptionContent> {
       override fun handle(content: NettyHttpContent) = handler(content)
     },
-    bodySize, autoRun, webSocketPath, readTimeout, writeTimeout, decodeType
+    bodySize, autoRun, webSocketPath, readTimeout, writeTimeout, decodeType, backlog, wrapWebSocketFrame
   )
 
   var decodeType: NettyHttpDecodeType = decodeType
@@ -84,6 +88,9 @@ class NettyHttpServer(
           pipeline.addLast("ws-$webSocketPath", WebSocketServerProtocolHandler(webSocketPath))
           pipeline.addLast("wsHandler-$webSocketPath", NettyWebSocketHandler(ch, handler))
         }
+        if (wrapWebSocketFrame && webSocketPath.iterator().hasNext()) {
+          pipeline.addLast(WebSocketFrameWrapper)
+        }
         pipeline.addLast("handle", httpHandler)
       }
     })