feat: complate socket.io base framework

Signed-off-by: MiaoWoo <admin@yumc.pw>

packages/websocket/src/server/client.ts

@@ -0,0 +1,27 @@
+import { EventEmitter } from 'events'
+
+const TextWebSocketFrame = Java.type('io.netty.handler.codec.http.websocketx.TextWebSocketFrame')
+
+export class NettyClient {
+    private event: EventEmitter
+
+    private _id: string;
+    private channel: any
+    constructor(channel: any) {
+        this.channel = channel;
+        this._id = channel.id();
+        this.event = new EventEmitter();
+    }
+    get id() {
+        return this._id;
+    }
+    on(event: string, callback: (...args: any[]) => void) {
+        this.event.on(event, callback);
+    }
+    emit(event: string, text: string) {
+        this.event.emit(event, text);
+    }
+    send(text: string) {
+        this.channel.writeAndFlush(new TextWebSocketFrame(text))
+    }
+}

packages/websocket/src/server/constants.ts

@@ -0,0 +1,13 @@
+export enum ServerEvent {
+    detect = 'detect',
+    connect = 'connect',
+    connection = 'connection',
+    message = 'message',
+    disconnect = 'disconnect'
+}
+
+export enum Keys {
+    Detect = "miao_detect",
+    Handler = "miaowebsocket",
+    Default = "DefaultChannelPipeline"
+}

packages/websocket/src/server/httprequest.ts

@@ -0,0 +1,66 @@
+import { HttpRequestHandlerAdapter } from '../netty'
+
+const DefaultHttpResponse = Java.type('io.netty.handler.codec.http.DefaultHttpResponse')
+const DefaultFullHttpResponse = Java.type('io.netty.handler.codec.http.DefaultFullHttpResponse')
+const HttpHeaders = Java.type('io.netty.handler.codec.http.HttpHeaders')
+const HttpVersion = Java.type('io.netty.handler.codec.http.HttpVersion')
+const HttpResponseStatus = Java.type('io.netty.handler.codec.http.HttpResponseStatus')
+const LastHttpContent = Java.type('io.netty.handler.codec.http.LastHttpContent')
+
+const File = Java.type('java.io.File')
+const Runnable = Java.type('java.lang.Runnable')
+const RandomAccessFile = Java.type('java.io.RandomAccessFile')
+const DefaultFileRegion = Java.type('io.netty.channel.DefaultFileRegion')
+const ChannelFutureListener = Java.type('io.netty.channel.ChannelFutureListener')
+
+export type HttpRequestConfig = {
+    root?: string;
+    ws?: string;
+}
+
+export class HttpRequestHandler extends HttpRequestHandlerAdapter {
+    private ws: string;
+    private root: string;
+    constructor(config: HttpRequestConfig = {
+        root: root + '/wwwroot',
+        ws: '/ws'
+    }) {
+        super()
+        this.root = config.root;
+        this.ws = config.ws;
+    }
+    channelRead0(ctx: any, request: any) {
+        if (request.getUri().startsWith(this.ws)) {
+            ctx.fireChannelRead(request.retain())
+        } else {
+            ctx.executor().execute(new Runnable({
+                run: () => {
+                    if (HttpHeaders.is100ContinueExpected(request)) {
+                        ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE))
+                    }
+                    let filename = request.getUri().split('?')[0].substr(1)
+                    let file = new File(this.root, filename || 'index.html')
+                    if (!file.exists() || !file.isFile()) {
+                        ctx.write(new DefaultHttpResponse(request.getProtocolVersion(), HttpResponseStatus.NOT_FOUND))
+                        ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE)
+                        return
+                    }
+                    let response = new DefaultHttpResponse(request.getProtocolVersion(), HttpResponseStatus.OK)
+                    response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/html charset=UTF-8")
+                    let raf = new RandomAccessFile(file, 'r')
+                    let keepAlive = HttpHeaders.isKeepAlive(request)
+                    if (keepAlive) {
+                        response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, file.length())
+                        response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE)
+                    }
+                    ctx.write(response)
+                    ctx.write(new DefaultFileRegion(raf.getChannel(), 0, raf.length()))
+                    let future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
+                    if (!keepAlive) {
+                        future.addListener(ChannelFutureListener.CLOSE)
+                    }
+                }
+            }))
+        }
+    }
+}

packages/websocket/src/server/index.ts

@@ -0,0 +1,53 @@
+import { EventEmitter } from 'events'
+
+import { ServerEvent, Keys } from './constants'
+import { WebSocketDetect } from './websocket_detect'
+import { WebSocketHandler } from './websocket_handler'
+import { NettyClient } from './client'
+
+interface NettyWebSocketServerOptions {
+    path?: string;
+}
+
+class NettyWebSocketServer {
+    private event: EventEmitter
+    private pipeline: any;
+    private allClients: { [key: string]: NettyClient };
+
+    constructor(pipeline: any, options: NettyWebSocketServerOptions) {
+        this.event = new EventEmitter();
+        this.allClients = {};
+        this.pipeline = pipeline;
+        let connectEvent = new EventEmitter();
+        this.pipeline.addFirst(Keys.Detect, new WebSocketDetect(connectEvent).getHandler())
+        connectEvent.on(ServerEvent.detect, (ctx, channel) => {
+            channel.pipeline().addFirst(Keys.Handler, new WebSocketHandler(connectEvent).getHandler())
+            ctx.fireChannelRead(channel)
+        })
+        connectEvent.on(ServerEvent.connect, (ctx) => {
+            let nettyClient = new NettyClient(ctx.channel());
+            this.allClients[nettyClient.id] = nettyClient;
+            this.event.emit(ServerEvent.connect, nettyClient);
+        })
+        connectEvent.on(ServerEvent.message, (ctx, msg) => {
+            let channel = ctx.channel();
+            this.allClients[channel.id()]?.emit(ServerEvent.message, msg.text())
+        })
+    }
+
+    disable() {
+        if (this.pipeline.names().contains(Keys.Detect)) {
+            this.pipeline.remove(Keys.Detect)
+        }
+    }
+
+    on(event: string, listener: (...args: any[]) => void) {
+        this.event.on(event, listener)
+    }
+}
+
+export {
+    NettyWebSocketServer,
+    ServerEvent,
+    NettyClient
+};

packages/websocket/src/server/text_websocket_frame.ts

@@ -0,0 +1,19 @@
+import { TextWebSocketFrameHandlerAdapter } from '../netty'
+import { EventEmitter } from 'events'
+import { ServerEvent } from './constants'
+
+export class TextWebSocketFrameHandler extends TextWebSocketFrameHandlerAdapter {
+    private event: EventEmitter;
+    constructor(event: EventEmitter) {
+        super()
+        this.event = event;
+    }
+    userEventTriggered(ctx: any, evt: any) {
+        if (evt == 'HANDSHAKE_COMPLETE') {
+            this.event.emit(ServerEvent.connect, ctx)
+        }
+    }
+    channelRead0(ctx: any, msg: any) {
+        this.event.emit(ServerEvent.message, ctx, msg)
+    }
+}

packages/websocket/src/server/websocket_detect.ts

@@ -0,0 +1,14 @@
+import { EventEmitter } from 'events'
+import { WebSocketHandlerAdapter } from "../netty"
+import { ServerEvent } from './constants'
+
+export class WebSocketDetect extends WebSocketHandlerAdapter {
+    private event: EventEmitter;
+    constructor(event: EventEmitter) {
+        super()
+        this.event = event;
+    }
+    channelRead(ctx: any, channel: any) {
+        this.event.emit(ServerEvent.detect, ctx, channel);
+    }
+}

packages/websocket/src/server/websocket_handler.ts

@@ -0,0 +1,42 @@
+import { EventEmitter } from 'events'
+
+import { Keys } from './constants'
+import { WebSocketHandlerAdapter } from "../netty"
+import { HttpRequestHandler } from './httprequest'
+import { TextWebSocketFrameHandler } from './text_websocket_frame'
+
+const CharsetUtil = Java.type('io.netty.util.CharsetUtil')
+
+const HttpServerCodec = Java.type('io.netty.handler.codec.http.HttpServerCodec')
+const ChunkedWriteHandler = Java.type('io.netty.handler.stream.ChunkedWriteHandler')
+const HttpObjectAggregator = Java.type('io.netty.handler.codec.http.HttpObjectAggregator')
+const WebSocketServerProtocolHandler = Java.type('io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler')
+
+export class WebSocketHandler extends WebSocketHandlerAdapter {
+    private event: EventEmitter;
+    constructor(event: EventEmitter) {
+        super()
+        this.event = event;
+    }
+    channelRead(ctx: any, msg: any) {
+        msg.markReaderIndex()
+        let message: string = msg.toString(CharsetUtil.UTF_8)
+        let channel = ctx.channel()
+        let pipeline = channel.pipeline()
+        if (message.indexOf('HTTP/1.1') > 0) {
+            pipeline.names().forEach(f => {
+                if (f == Keys.Handler || f.indexOf(Keys.Default) > -1) { return }
+                pipeline.remove(f)
+            })
+            pipeline.addLast('http', new HttpServerCodec())
+            pipeline.addLast('chunk', new ChunkedWriteHandler())
+            pipeline.addLast('httpobj', new HttpObjectAggregator(64 * 1024))
+            pipeline.addLast('http_request', new HttpRequestHandler().getHandler())
+            pipeline.addLast('websocket', new WebSocketServerProtocolHandler("/ws", true))
+            pipeline.addLast('websocket_handler', new TextWebSocketFrameHandler(this.event).getHandler())
+        }
+        pipeline.remove(Keys.Handler)
+        msg.resetReaderIndex()
+        ctx.fireChannelRead(msg)
+    }
+}