circlecloud
/
ms
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

feat: support tomcat websocket 

Signed-off-by: MiaoWoo <admin@yumc.pw>
backup
MiaoWoo 2020-05-26 15:45:40 +08:00
parent 4e1a77cc50
commit c37a7cb772
14 changed files with 185 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
packages/websocket/src/server/client.ts
View File

@ -1,5 +1,5 @@
import { EventEmitter } from 'events'
import { SocketIO } from 'socket-io/interfaces';
import { SocketIO } from '../socket-io/interfaces';
import { AttributeKeys } from './constants';
const TextWebSocketFrame = Java.type('io.netty.handler.codec.http.websocketx.TextWebSocketFrame')

9
packages/websocket/src/server/constants.ts
View File

@ -1,12 +1,3 @@
export enum ServerEvent {
detect = 'detect',
connect = 'connect',
connection = 'connection',
message = 'message',
error = 'error',
disconnect = 'disconnect'
}
const AttributeKey = Java.type('io.netty.util.AttributeKey');
export enum Keys {

6
packages/websocket/src/server/index.ts
View File

@ -1,8 +1,10 @@
import { EventEmitter } from 'events'
import { NettyClient } from './client'
import { ServerOptions } from '../socket-io'
import { ServerEvent, Keys } from './constants'
import { ServerEvent } from '../socket-io/constants'
import { NettyClient } from './client'
import { Keys } from './constants'
import { WebSocketDetect } from './websocket_detect'
import { WebSocketHandler } from './websocket_handler'

2
packages/websocket/src/server/text_websocket_frame.ts
View File

@ -1,6 +1,6 @@
import { EventEmitter } from 'events'
import { ServerEvent } from './constants'
import { ServerOptions } from '../socket-io';
import { ServerEvent } from '../socket-io/constants'
import { TextWebSocketFrameHandlerAdapter } from '../netty'
export class TextWebSocketFrameHandler extends TextWebSocketFrameHandlerAdapter {

2
packages/websocket/src/server/websocket_detect.ts
View File

@ -1,6 +1,6 @@
import { EventEmitter } from 'events'
import { WebSocketHandlerAdapter } from "../netty"
import { ServerEvent } from './constants'
import { ServerEvent } from '../socket-io/constants'
export class WebSocketDetect extends WebSocketHandlerAdapter {
private event: EventEmitter;

4
packages/websocket/src/server/websocket_handler.ts
View File

@ -1,5 +1,7 @@
import { ServerOptions } from '../socket-io'
import { Keys, ServerEvent } from './constants'
import { ServerEvent } from '../socket-io/constants'
import { Keys } from './constants'
import { HttpRequestHandler } from './httprequest'
import { WebSocketHandlerAdapter } from "../netty"
import { TextWebSocketFrameHandler } from './text_websocket_frame'

24
packages/websocket/src/socket-io/client.ts
View File

@ -1,30 +1,34 @@
import { EventEmitter } from 'events'
import { Parser } from './parser'
import { Packet } from './packet';
import { NettyClient } from '../server';
import { SocketIO } from './interfaces'
import { Server, Socket } from './index';
import { PacketTypes, SubPacketTypes } from './types';
import { ServerEvent } from './constants';
const parser = new Parser();
export class Client extends EventEmitter implements SocketIO.Client {
id: string;
server: Server;
conn: NettyClient;
conn: SocketIO.EngineSocket;
request: any;
sockets: { [id: string]: Socket; };
nsps: { [nsp: string]: SocketIO.Socket; };
connectBuffer: any;
constructor(server: Server, nettyClient: NettyClient) {
constructor(server: Server, engine: SocketIO.EngineSocket) {
super();
this.server = server;
this.conn = nettyClient;
this.conn = engine;
this.id = this.conn.id + '';
this.request = nettyClient.request;
this.request = engine.request;
this.sockets = {};
this.nsps = {};
this.conn.on(ServerEvent.disconnect, (reason) => {
this.onclose(reason)
})
}
connect(name, query) {
if (this.server.nsps[name]) {
@ -79,11 +83,11 @@ export class Client extends EventEmitter implements SocketIO.Client {
// this.decoder.destroy(); // clean up decoder
}
disconnect() {
// if ('open' == this.conn.readyState) {
// debug('forcing transport close');
this.conn.close();
this.onclose('forced server close');
// }
if ('open' == this.conn.readyState) {
// debug('forcing transport close');
this.conn.close();
this.onclose('forced server close');
}
}
remove(socket: Socket) {
if (this.sockets.hasOwnProperty(socket.id)) {

8
packages/websocket/src/socket-io/constants.ts Normal file
View File

@ -0,0 +1,8 @@
export enum ServerEvent {
detect = 'detect',
connect = 'connect',
connection = 'connection',
message = 'message',
error = 'error',
disconnect = 'disconnect'
}

48
packages/websocket/src/socket-io/index.ts
View File

@ -1,8 +1,6 @@
import { EventEmitter } from 'events'
import { NettyWebSocketServer, NettyClient } from '../server'
import { ServerEvent } from '../server/constants';
import { ServerEvent } from './constants';
import { Namespace } from './namespace';
import { Client } from './client';
import { SocketIO } from './interfaces'
@ -17,8 +15,12 @@ interface ServerOptions extends SocketIO.ServerOptions {
root?: string;
}
interface WebSocketServer extends EventEmitter {
close(): void
}
class Server implements SocketIO.Server {
private nettyServer: NettyWebSocketServer;
private websocketServer: WebSocketServer;
private allClients: { [key: string]: Client };
engine: { ws: any; };
@ -31,17 +33,24 @@ class Server implements SocketIO.Server {
_adapter: Adapter;
options: ServerOptions;
constructor(pipeline: any, options: ServerOptions) {
if (!pipeline) { throw new Error('Netty Pipeline can\'t be undefiend!') }
constructor(instance: any, options: ServerOptions) {
if (!instance) { throw new Error('instance can\'t be undefiend!') }
this.allClients = {};
this.nsps = {};
this.sockets = new Namespace('/', this);
this.nsps['/'] = this.sockets;
this.initNettyServer(pipeline, Object.assign({
event: new EventEmitter(),
path: '/socket.io',
root: root + '/wwwroot'
}, options));
if (instance.class.name.startsWith('io.netty.channel')) {
let { NettyWebSocketServer } = require("../server")
this.websocketServer = new NettyWebSocketServer(instance, Object.assign({
event: new EventEmitter(),
path: '/socket.io',
root: root + '/wwwroot'
}, options));
} else {
let { TomcatWebSocketServer } = require("../tomcat/server")
this.websocketServer = new TomcatWebSocketServer(instance, options);
}
this.initServer()
}
checkRequest(req: any, fn: (err: any, success: boolean) => void): void {
@ -114,7 +123,7 @@ class Server implements SocketIO.Server {
for (let socket in this.sockets.sockets) {
this.sockets.sockets[socket].onclose()
}
this.nettyServer.close();
this.websocketServer.close();
}
on(event: "connection", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace;
on(event: "connect", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace;
@ -152,17 +161,16 @@ class Server implements SocketIO.Server {
fn(false);
};
private initNettyServer(pipeline, options) {
this.nettyServer = new NettyWebSocketServer(pipeline, options);
this.nettyServer.on(ServerEvent.connect, (nettyClient: NettyClient) => {
let client = new Client(this, nettyClient);
private initServer() {
this.websocketServer.on(ServerEvent.connect, (socket: SocketIO.EngineSocket) => {
let client = new Client(this, socket);
this.onconnection(client);
})
this.nettyServer.on(ServerEvent.message, (nettyClient: NettyClient, text) => {
this.processPacket(this.parser.decode(text), this.allClients[nettyClient.id]);
this.websocketServer.on(ServerEvent.message, (socket: SocketIO.EngineSocket, text) => {
this.processPacket(this.parser.decode(text), this.allClients[socket.id]);
})
this.nettyServer.on(ServerEvent.error, (nettyClient: NettyClient, cause) => {
console.error(`Client ${nettyClient.id} cause error: ` + cause)
this.websocketServer.on(ServerEvent.error, (socket: SocketIO.EngineSocket, cause) => {
console.error(`Client ${socket.id} cause error: ` + cause)
console.ex(cause)
})
}

10
packages/websocket/src/socket-io/interfaces.ts
View File

@ -830,5 +830,15 @@ export declare namespace SocketIO {
* (Transport): transport reference
*/
transport: any;
/**
* send
*/
send(text: string);
/**
* close
*/
close();
}
}

2
packages/websocket/src/socket-io/namespace.ts
View File

@ -2,7 +2,7 @@ import { EventEmitter } from 'events'
import { Client } from './client'
import { SocketIO } from './interfaces';
import { ServerEvent } from '../server';
import { ServerEvent } from './constants';
import { Socket } from './socket';
import { Adapter } from './adapter';
import { Server } from './index'

44
packages/websocket/src/tomcat/client.ts Normal file
View File

@ -0,0 +1,44 @@
import { EventEmitter } from 'events'
import { SocketIO } from '../socket-io/interfaces';
export class TomcatClient extends EventEmitter implements SocketIO.EngineSocket {
private _id: string;
private session: any
server: any;
readyState: string;
remoteAddress: string;
upgraded: boolean;
request: any;
transport: any;
constructor(server: any, session: any) {
super();
this.server = server;
this.readyState = 'open';
this.remoteAddress = session + ''
this.upgraded = true;
this.request = {
uri: () => {
return session.getRequestURI() + ''
},
headers: () => {
return []
}
};
this.transport = null;
this.session = session;
this._id = session.getId();
}
get id() {
return this._id;
}
send(text: string) {
this.session.getBasicRemote().sendText(text)
}
close() {
this.session.close();
}
}

1
packages/websocket/src/tomcat/constants.ts Normal file
View File

@ -0,0 +1 @@
export const ProxyBeanName = "webSocketServerProxy"

69
packages/websocket/src/tomcat/server.ts Normal file
View File

@ -0,0 +1,69 @@
import { EventEmitter } from 'events'
import { ServerOptions } from '../socket-io'
import { ServerEvent } from '../socket-io/constants'
import { SocketIO } from '../socket-io/interfaces'
import { ProxyBeanName } from './constants'
import { TomcatClient } from './client'
const WebSocketServerProxy = Java.type("com.sixi.framework.scriptservice.websocket.WebSocketServerProxy")
const ThreadPoolExecutor = Java.type('java.util.concurrent.ThreadPoolExecutor')
const ThreadPoolTaskExecutor = Java.type('org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor')
interface TomcatWebSocketSession {
getId: () => number
}
class TomcatWebSocketServer extends EventEmitter {
private beanFactory: any
private executor: any
private allClients: { [key: string]: SocketIO.EngineSocket }
constructor(beanFactory: any, options: ServerOptions) {
super()
this.allClients = {}
this.beanFactory = beanFactory
this.initThreadPool()
try { this.beanFactory.destroySingleton(ProxyBeanName) } catch (error) { }
let NashornWebSocketServerProxy = Java.extend(WebSocketServerProxy, {
onOpen: (session: TomcatWebSocketSession) => {
let tomcatClient = new TomcatClient(this, session)
this.allClients[tomcatClient.id] = tomcatClient
this.emit(ServerEvent.connect, tomcatClient)
},
onMessage: (message: any, session: TomcatWebSocketSession) => {
this.executor.execute(() => {
this.emit(ServerEvent.message, this.allClients[session.getId()], message)
})
},
onClose: (session: TomcatWebSocketSession, reason: any) => {
this.emit(ServerEvent.disconnect, this.allClients[session.getId()], reason)
},
onError: (session: TomcatWebSocketSession, error: any) => {
this.emit(ServerEvent.error, this.allClients[session.getId()], error)
},
})
this.beanFactory.registerSingleton(ProxyBeanName, new NashornWebSocketServerProxy())
}
private initThreadPool() {
this.executor = new ThreadPoolTaskExecutor()
this.executor.setCorePoolSize(10)
this.executor.setMaxPoolSize(100)
this.executor.setQueueCapacity(500)
this.executor.setKeepAliveSeconds(60)
this.executor.setThreadNamePrefix("@ccms/websocket-")
this.executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy())
this.executor.initialize()
}
close() {
Object.values(this.allClients).forEach(client => client.close())
this.beanFactory.destroySingleton(ProxyBeanName)
this.executor.shutdown()
}
}
export {
TomcatWebSocketServer,
ServerEvent,
TomcatClient
}