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

fix: boardcast error when socket send packet

Browse Source 
Signed-off-by: MiaoWoo <admin@yumc.pw>
This commit is contained in:
MiaoWoo 2020-06-30 14:02:18 +08:00
parent f4aa568f0c
commit 678f4ca8a4
3 changed files with 247 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,18 +1,18 @@
import { EventEmitter } from 'events' import { EventEmitter } from 'events'
import { ServerEvent } from './constants'; import { ServerEvent } from './constants'
import { Namespace } from './namespace'; import { Namespace } from './namespace'
import { Client } from './client'; import { Client } from './client'
import { SocketIO } from './interfaces' import { SocketIO } from './interfaces'
import { Parser } from './parser' import { Parser } from './parser'
import { PacketTypes, SubPacketTypes } from './types'; import { PacketTypes, SubPacketTypes } from './types'
import { Packet } from './packet'; import { Packet } from './packet'
import { Socket } from './socket'; import { Socket } from './socket'
import { Adapter } from './adapter'; import { Adapter } from './adapter'
interface ServerOptions extends SocketIO.ServerOptions { interface ServerOptions extends SocketIO.ServerOptions {
event?: EventEmitter; event?: EventEmitter
root?: string; root?: string
} }
interface WebSocketServer extends EventEmitter { interface WebSocketServer extends EventEmitter {
@ -20,87 +20,87 @@ interface WebSocketServer extends EventEmitter {
} }
class Server implements SocketIO.Server { class Server implements SocketIO.Server {
private websocketServer: WebSocketServer; private websocketServer: WebSocketServer
private allClients: { [key: string]: Client }; private allClients: { [key: string]: Client }
engine: { ws: any; }; engine: { ws: any }
nsps: { [namespace: string]: Namespace; }; nsps: { [namespace: string]: Namespace }
sockets: Namespace; sockets: Namespace
json: SocketIO.Server; json: SocketIO.Server
volatile: SocketIO.Server; volatile: SocketIO.Server
local: SocketIO.Server; local: SocketIO.Server
parser = new Parser(); parser = new Parser();
_adapter: Adapter; _adapter: Adapter
options: ServerOptions; options: ServerOptions
constructor(instance: any, options: ServerOptions) { constructor(instance: any, options: ServerOptions) {
if (!instance) { throw new Error('instance can\'t be undefiend!') } if (!instance) { throw new Error('instance can\'t be undefiend!') }
this.allClients = {}; this.allClients = {}
this.nsps = {}; this.nsps = {}
this.sockets = new Namespace('/', this); this.sockets = new Namespace('/', this)
this.nsps['/'] = this.sockets; this.nsps['/'] = this.sockets
if (instance.class.name.startsWith('io.netty.channel')) { if (instance.class.name.startsWith('io.netty.channel')) {
let { NettyWebSocketServer } = require("../server") let { NettyWebSocketServer } = require("../server")
this.websocketServer = new NettyWebSocketServer(instance, Object.assign({ this.websocketServer = new NettyWebSocketServer(instance, Object.assign({
event: new EventEmitter(), event: new EventEmitter(),
path: '/socket.io', path: '/socket.io',
root: root + '/wwwroot' root: root + '/wwwroot'
}, options)); }, options))
} else { } else {
let { TomcatWebSocketServer } = require("../tomcat/server") let { TomcatWebSocketServer } = require("../tomcat/server")
this.websocketServer = new TomcatWebSocketServer(instance, options); this.websocketServer = new TomcatWebSocketServer(instance, options)
} }
this.initServer() this.initServer()
} }
checkRequest(req: any, fn: (err: any, success: boolean) => void): void { checkRequest(req: any, fn: (err: any, success: boolean) => void): void {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
serveClient(): boolean; serveClient(): boolean
serveClient(v: boolean): SocketIO.Server; serveClient(v: boolean): SocketIO.Server
serveClient(v?: any): boolean | SocketIO.Server { serveClient(v?: any): boolean | SocketIO.Server {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
path(): string; path(): string
path(v: string): SocketIO.Server; path(v: string): SocketIO.Server
path(v?: any): string | SocketIO.Server { path(v?: any): string | SocketIO.Server {
if (!arguments.length) return this.options.path; if (!arguments.length) return this.options.path
this.options.path = v.replace(/\/$/, ''); this.options.path = v.replace(/\/$/, '')
return this; return this
} }
adapter(): Adapter; adapter(): Adapter
adapter(v: any): SocketIO.Server; adapter(v: any): SocketIO.Server
adapter(v?: any): Adapter | SocketIO.Server { adapter(v?: any): Adapter | SocketIO.Server {
if (!arguments.length) return this._adapter; if (!arguments.length) return this._adapter
this._adapter = v; this._adapter = v
for (var i in this.nsps) { for (var i in this.nsps) {
if (this.nsps.hasOwnProperty(i)) { if (this.nsps.hasOwnProperty(i)) {
this.nsps[i].initAdapter(); this.nsps[i].initAdapter()
} }
} }
return this; return this
} }
origins(): string | string[]; origins(): string | string[]
origins(v: string | string[]): SocketIO.Server; origins(v: string | string[]): SocketIO.Server
origins(fn: (origin: string, callback: (error: string, success: boolean) => void) => void): SocketIO.Server; origins(fn: (origin: string, callback: (error: string, success: boolean) => void) => void): SocketIO.Server
origins(fn?: any): string | string[] | SocketIO.Server { origins(fn?: any): string | string[] | SocketIO.Server {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
attach(srv: any, opts?: SocketIO.ServerOptions): SocketIO.Server; attach(srv: any, opts?: SocketIO.ServerOptions): SocketIO.Server
attach(port: number, opts?: SocketIO.ServerOptions): SocketIO.Server; attach(port: number, opts?: SocketIO.ServerOptions): SocketIO.Server
attach(port: any, opts?: any): SocketIO.Server { attach(port: any, opts?: any): SocketIO.Server {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
listen(srv: any, opts?: SocketIO.ServerOptions): SocketIO.Server; listen(srv: any, opts?: SocketIO.ServerOptions): SocketIO.Server
listen(port: number, opts?: SocketIO.ServerOptions): SocketIO.Server; listen(port: number, opts?: SocketIO.ServerOptions): SocketIO.Server
listen(port: any, opts?: any): SocketIO.Server { listen(port: any, opts?: any): SocketIO.Server {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
bind(srv: any): SocketIO.Server { bind(srv: any): SocketIO.Server {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
onconnection(socket: Client): SocketIO.Server { onconnection(socket: Client): SocketIO.Server {
this.allClients[socket.id] = socket; this.allClients[socket.id] = socket
socket.packet({ socket.packet({
type: PacketTypes.OPEN, type: PacketTypes.OPEN,
data: { data: {
@ -110,68 +110,75 @@ class Server implements SocketIO.Server {
pingTimeout: 5000 pingTimeout: 5000
} }
}) })
this.sockets.add(socket); this.sockets.add(socket)
return this; return this
} }
of(nsp: string): Namespace { of(nsp: string): Namespace {
if (!this.nsps[nsp]) { if (!this.nsps[nsp]) {
this.nsps[nsp] = new Namespace(nsp, this); this.nsps[nsp] = new Namespace(nsp, this)
} }
return this.nsps[nsp]; return this.nsps[nsp]
} }
close(fn?: () => void): void { close(fn?: () => void): void {
for (let socket in this.sockets.sockets) { for (let socket in this.sockets.sockets) {
this.sockets.sockets[socket].onclose() this.sockets.sockets[socket].onclose()
} }
this.websocketServer.close(); this.websocketServer.close()
} }
on(event: "connection", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace; on(event: "connection", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace
on(event: "connect", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace; on(event: "connect", listener: (socket: SocketIO.Socket) => void): SocketIO.Namespace
on(event: string, listener: Function): SocketIO.Namespace; on(event: string, listener: Function): SocketIO.Namespace
on(event: any, listener: any): SocketIO.Namespace { on(event: any, listener: any): SocketIO.Namespace {
return this.sockets.on(event, listener); return this.sockets.on(event, listener)
} }
to(room: string): SocketIO.Namespace { to(room: string): SocketIO.Namespace {
return this.sockets.to(room); return this.sockets.to(room)
} }
in(room: string): SocketIO.Namespace { in(room: string): SocketIO.Namespace {
return this.sockets.in(room); return this.sockets.in(room)
} }
use(fn: (socket: SocketIO.Socket, fn: (err?: any) => void) => void): SocketIO.Namespace { use(fn: (socket: SocketIO.Socket, fn: (err?: any) => void) => void): SocketIO.Namespace {
return this.sockets.use(fn); return this.sockets.use(fn)
} }
emit(event: string, ...args: any[]): SocketIO.Namespace { emit(event: string, ...args: any[]): SocketIO.Namespace {
// @ts-ignore // @ts-ignore
return this.sockets.emit(event, ...args); return this.sockets.emit(event, ...args)
} }
send(...args: any[]): SocketIO.Namespace { send(...args: any[]): SocketIO.Namespace {
return this.sockets.send(...args); return this.sockets.send(...args)
} }
write(...args: any[]): SocketIO.Namespace { write(...args: any[]): SocketIO.Namespace {
return this.sockets.write(...args); return this.sockets.write(...args)
} }
clients(...args: any[]): SocketIO.Namespace { clients(...args: any[]): SocketIO.Namespace {
return this.sockets.clients(args[0]); return this.sockets.clients(args[0])
} }
compress(...args: any[]): SocketIO.Namespace { compress(...args: any[]): SocketIO.Namespace {
return this.sockets.compress(args[0]) return this.sockets.compress(args[0])
} }
// =============================== // ===============================
checkNamespace(name, query, fn) { checkNamespace(name, query, fn) {
fn(false); fn(false)
}; };
private initServer() { private initServer() {
this.websocketServer.on(ServerEvent.connect, (socket: SocketIO.EngineSocket) => { this.websocketServer.on(ServerEvent.connect, (socket: SocketIO.EngineSocket) => {
let client = new Client(this, socket); let client = new Client(this, socket)
this.onconnection(client); this.onconnection(client)
}) })
this.websocketServer.on(ServerEvent.message, (socket: SocketIO.EngineSocket, text) => { this.websocketServer.on(ServerEvent.message, (socket: SocketIO.EngineSocket, text) => {
this.processPacket(this.parser.decode(text), this.allClients[socket.id]); this.processPacket(this.parser.decode(text), this.allClients[socket.id])
})
this.websocketServer.on(ServerEvent.disconnect, (socket: SocketIO.EngineSocket, reason) => {
this.allClients[socket.id].onclose(reason)
delete this.allClients[socket.id]
}) })
this.websocketServer.on(ServerEvent.error, (socket: SocketIO.EngineSocket, cause) => { this.websocketServer.on(ServerEvent.error, (socket: SocketIO.EngineSocket, cause) => {
console.error(`Client ${socket.id} cause error: ` + cause) if (socket.listeners(ServerEvent.error).length) {
console.ex(cause) socket.emit(ServerEvent.error, cause)
} else {
console.error(`client ${socket.id} cause error: ${cause}`)
}
}) })
} }
@ -182,15 +189,15 @@ class Server implements SocketIO.Server {
type: PacketTypes.PONG, type: PacketTypes.PONG,
data: packet.data data: packet.data
}) })
break; break
case PacketTypes.UPGRADE: case PacketTypes.UPGRADE:
break; break
case PacketTypes.MESSAGE: case PacketTypes.MESSAGE:
this.processSubPacket(packet, client); this.processSubPacket(packet, client)
break; break
case PacketTypes.CLOSE: case PacketTypes.CLOSE:
client.onclose() client.onclose()
break; break
} }
} }
@ -201,11 +208,11 @@ class Server implements SocketIO.Server {
type: PacketTypes.MESSAGE, type: PacketTypes.MESSAGE,
sub_type: SubPacketTypes.ERROR, sub_type: SubPacketTypes.ERROR,
data: 'not support dynamic namespace: ' + packet.nsp data: 'not support dynamic namespace: ' + packet.nsp
}); })
client.disconnect(); client.disconnect()
return; return
} }
namespace.process(packet, client); namespace.process(packet, client)
} }
} }
export { export {

253
packages/websocket/src/socket-io/socket.ts
View File

@ -1,28 +1,29 @@
import { EventEmitter } from 'events' import { EventEmitter } from 'events'
import { SocketIO } from "./interfaces"; import { SocketIO } from "./interfaces"
import { Packet } from './packet'; import { Packet } from './packet'
import { PacketTypes, SubPacketTypes } from './types'; import { PacketTypes, SubPacketTypes } from './types'
import { Client } from './client'; import { Client } from './client'
import { Namespace } from './namespace'; import { Namespace } from './namespace'
import * as querystring from 'querystring' import * as querystring from 'querystring'
import { ServerEvent } from './constants'
export class Socket extends EventEmitter implements SocketIO.Socket { export class Socket extends EventEmitter implements SocketIO.Socket {
nsp: Namespace; nsp: Namespace
server: SocketIO.Server; server: SocketIO.Server
adapter: SocketIO.Adapter; adapter: SocketIO.Adapter
id: string; id: string
request: any; request: any
client: Client; client: Client
conn: SocketIO.EngineSocket; conn: SocketIO.EngineSocket
rooms: { [id: string]: string; }; rooms: { [id: string]: string }
acks: { [id: string]: Function; }; acks: { [id: string]: Function }
connected: boolean; connected: boolean
disconnected: boolean; disconnected: boolean
handshake: SocketIO.Handshake; handshake: SocketIO.Handshake
fns: any[]; fns: any[]
flags: { [key: string]: boolean }; flags: { [key: string]: boolean }
_rooms: string[]; _rooms: string[]
private events = [ private events = [
'error', 'error',
@ -34,113 +35,113 @@ export class Socket extends EventEmitter implements SocketIO.Socket {
] ]
constructor(nsp: Namespace, client: Client, query = {}) { constructor(nsp: Namespace, client: Client, query = {}) {
super(); super()
this.nsp = nsp; this.nsp = nsp
this.server = nsp.server; this.server = nsp.server
this.adapter = this.nsp.adapter; this.adapter = this.nsp.adapter
this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id; this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id
this.client = client; this.client = client
this.request = client.request; this.request = client.request
this.conn = client.conn; this.conn = client.conn
this.rooms = {}; this.rooms = {}
this.acks = {}; this.acks = {}
this.connected = true; this.connected = true
this.disconnected = false; this.disconnected = false
this.handshake = this.buildHandshake(query); this.handshake = this.buildHandshake(query)
this.fns = []; this.fns = []
this.flags = {}; this.flags = {}
this._rooms = []; this._rooms = []
} }
get json() { get json() {
this.flags.json = true; this.flags.json = true
return this return this
} }
get volatile() { get volatile() {
this.flags.volatile = true; this.flags.volatile = true
return this return this
} }
get broadcast() { get broadcast() {
this.flags.broadcast = true; this.flags.broadcast = true
return this return this
} }
get local() { get local() {
this.flags.local = true; this.flags.local = true
return this return this
} }
to(room: string): SocketIO.Socket { to(room: string): SocketIO.Socket {
if (!~this._rooms.indexOf(room)) this._rooms.push(room); if (!~this._rooms.indexOf(room)) this._rooms.push(room)
return this; return this
} }
in(room: string): SocketIO.Socket { in(room: string): SocketIO.Socket {
return this.to(room); return this.to(room)
} }
use(fn: (packet: SocketIO.Packet, next: (err?: any) => void) => void): SocketIO.Socket { use(fn: (packet: SocketIO.Packet, next: (err?: any) => void) => void): SocketIO.Socket {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
send(...args: any[]): SocketIO.Socket { send(...args: any[]): SocketIO.Socket {
this.emit("message", ...args) this.emit("message", ...args)
return this; return this
} }
write(...args: any[]): SocketIO.Socket { write(...args: any[]): SocketIO.Socket {
return this.send(...args); return this.send(...args)
} }
join(rooms: string | string[], fn?: (err?: any) => void): SocketIO.Socket { join(rooms: string | string[], fn?: (err?: any) => void): SocketIO.Socket {
if (!Array.isArray(rooms)) { if (!Array.isArray(rooms)) {
rooms = [rooms]; rooms = [rooms]
} }
rooms = rooms.filter((room) => { rooms = rooms.filter((room) => {
return !this.rooms.hasOwnProperty(room); return !this.rooms.hasOwnProperty(room)
}); })
if (!rooms.length) { if (!rooms.length) {
fn && fn(null); fn && fn(null)
return this; return this
} }
this.adapter.addAll(this.id, rooms, (err) => { this.adapter.addAll(this.id, rooms, (err) => {
if (err) return fn && fn(err); if (err) return fn && fn(err);
// debug('joined room %s', rooms); // debug('joined room %s', rooms);
(rooms as Array<string>).forEach((room) => { (rooms as Array<string>).forEach((room) => {
this.rooms[room] = room; this.rooms[room] = room
}); })
fn && fn(null); fn && fn(null)
}); })
return this; return this
} }
leave(name: string, fn?: Function): SocketIO.Socket { leave(name: string, fn?: Function): SocketIO.Socket {
delete this.rooms[name]; delete this.rooms[name]
fn && fn(null) fn && fn(null)
return this; return this
} }
leaveAll(): void { leaveAll(): void {
this.adapter.delAll(this.id); this.adapter.delAll(this.id)
this.rooms = {}; this.rooms = {}
} }
disconnect(close?: boolean): SocketIO.Socket { disconnect(close?: boolean): SocketIO.Socket {
if (!this.connected) return this; if (!this.connected) return this
if (close) { if (close) {
this.client.disconnect(); this.client.disconnect()
} else { } else {
this.packet({ type: PacketTypes.MESSAGE, sub_type: SubPacketTypes.DISCONNECT }); this.packet({ type: PacketTypes.MESSAGE, sub_type: SubPacketTypes.DISCONNECT })
this.onclose('server namespace disconnect'); this.onclose('server namespace disconnect')
} }
return this; return this
} }
compress(compress: boolean): SocketIO.Socket { compress(compress: boolean): SocketIO.Socket {
throw new Error("Method not implemented."); throw new Error("Method not implemented.")
} }
error(err: any): void { error(err: any): void {
this.packet({ type: PacketTypes.MESSAGE, sub_type: SubPacketTypes.ERROR, data: err }); this.packet({ type: PacketTypes.MESSAGE, sub_type: SubPacketTypes.ERROR, data: err })
} }
// ========================================== // ==========================================
buildHandshake(query): SocketIO.Handshake { buildHandshake(query): SocketIO.Handshake {
let requestUri = this.request.uri(); let requestUri = this.request.uri()
let headers = {}; let headers = {}
let nativeHeaders = this.request.headers(); let nativeHeaders = this.request.headers()
nativeHeaders.forEach(function (header) { nativeHeaders.forEach(function (header) {
headers[header.getKey()] = header.getValue(); headers[header.getKey()] = header.getValue()
}) })
return { return {
headers: headers, headers: headers,
@ -155,9 +156,9 @@ export class Socket extends EventEmitter implements SocketIO.Socket {
} }
emit(event: string, ...args: any[]): boolean { emit(event: string, ...args: any[]): boolean {
if (~this.events.indexOf(event)) { if (~this.events.indexOf(event)) {
super.emit(event, ...args); super.emit(event, ...args)
// @ts-ignore // @ts-ignore
return this; return this
} }
let packet: Packet = { let packet: Packet = {
@ -170,44 +171,48 @@ export class Socket extends EventEmitter implements SocketIO.Socket {
// access last argument to see if it's an ACK callback // access last argument to see if it's an ACK callback
if (typeof args[args.length - 1] === 'function') { if (typeof args[args.length - 1] === 'function') {
if (this._rooms.length || this.flags.broadcast) { if (this._rooms.length || this.flags.broadcast) {
throw new Error('Callbacks are not supported when broadcasting'); throw new Error('Callbacks are not supported when broadcasting')
} }
// debug('emitting packet with ack id %d', this.nsp.ids); // debug('emitting packet with ack id %d', this.nsp.ids);
this.acks[this.nsp.ids] = args.pop(); this.acks[this.nsp.ids] = args.pop()
packet.id = this.nsp.ids++; packet.id = this.nsp.ids++
} }
let rooms = this._rooms.slice(0); let rooms = this._rooms.slice(0)
let flags = Object.assign({}, this.flags); let flags = Object.assign({}, this.flags)
// reset flags // reset flags
this._rooms = []; this._rooms = []
this.flags = {}; this.flags = {}
if (rooms.length || flags.broadcast) { if (rooms.length || flags.broadcast) {
this.adapter.broadcast(packet, { this.adapter.broadcast(packet, {
except: [this.id], except: [this.id],
rooms: rooms, rooms: rooms,
flags: flags flags: flags
}); })
} else { } else {
// dispatch packet // dispatch packet
this.packet(packet, flags); this.packet(packet, flags)
} }
// @ts-ignore // @ts-ignore
return this; return this
} }
packet(packet: Packet, opts: any = { preEncoded: false }) { packet(packet: Packet, opts: any = { preEncoded: false }) {
if (!opts.preEncoded) { if (!opts.preEncoded) {
packet.nsp = this.nsp.name; packet.nsp = this.nsp.name
opts.compress = false !== opts.compress; opts.compress = false !== opts.compress
}
try {
this.client.packet(packet, opts)
} catch (error) {
this.onerror(error)
} }
this.client.packet(packet, opts);
} }
onconnect() { onconnect() {
this.nsp.connected[this.id] = this; this.nsp.connected[this.id] = this
this.client.sockets[this.id] = this; this.client.sockets[this.id] = this
this.join(this.id); this.join(this.id)
// let skip = this.nsp.name === '/' && this.nsp.fns.length === 0; // let skip = this.nsp.name === '/' && this.nsp.fns.length === 0;
// if (skip) { // if (skip) {
// debug('packet already sent in initial handshake'); // debug('packet already sent in initial handshake');
@ -215,54 +220,54 @@ export class Socket extends EventEmitter implements SocketIO.Socket {
this.packet({ this.packet({
type: PacketTypes.MESSAGE, type: PacketTypes.MESSAGE,
sub_type: SubPacketTypes.CONNECT sub_type: SubPacketTypes.CONNECT
}); })
// } // }
} }
onclose(reason?: string) { onclose(reason?: string) {
if (!this.connected) return this; if (!this.connected) return this
// debug('closing socket - reason %s', reason); // debug('closing socket - reason %s', reason);
this.emit('disconnecting', reason); this.emit('disconnecting', reason)
this.leaveAll(); this.leaveAll()
this.nsp.remove(this); this.nsp.remove(this)
this.client.remove(this); this.client.remove(this)
this.connected = false; this.connected = false
this.disconnected = true; this.disconnected = true
delete this.nsp.connected[this.id]; delete this.nsp.connected[this.id]
this.emit('disconnect', reason); this.emit('disconnect', reason)
} }
onpacket(packet: Packet) { onpacket(packet: Packet) {
switch (packet.sub_type) { switch (packet.sub_type) {
// 2 // 2
case SubPacketTypes.EVENT: case SubPacketTypes.EVENT:
this.onevent(packet); this.onevent(packet)
break; break
// 5 // 5
case SubPacketTypes.BINARY_EVENT: case SubPacketTypes.BINARY_EVENT:
this.onevent(packet); this.onevent(packet)
break; break
// 3 // 3
case SubPacketTypes.ACK: case SubPacketTypes.ACK:
this.onack(packet); this.onack(packet)
break; break
// 6 // 6
case SubPacketTypes.BINARY_ACK: case SubPacketTypes.BINARY_ACK:
this.onack(packet); this.onack(packet)
break; break
// 1 // 1
case SubPacketTypes.DISCONNECT: case SubPacketTypes.DISCONNECT:
this.ondisconnect(); this.ondisconnect()
break; break
// 4 // 4
case SubPacketTypes.ERROR: case SubPacketTypes.ERROR:
this.onerror(new Error(packet.data)); this.onerror(new Error(packet.data))
} }
} }
onerror(err: Error) { onerror(err: Error) {
if (this.listeners('error').length) { if (this.listeners('error').length) {
this.emit('error', err); this.emit('error', err)
} else { } else {
console.error('Missing error handler on `socket`.'); console.error('Missing error handler on `socket`.')
console.error(err.stack); console.error(err.stack)
} }
} }
ondisconnect() { ondisconnect() {
@ -271,39 +276,39 @@ export class Socket extends EventEmitter implements SocketIO.Socket {
onevent(packet: Packet) { onevent(packet: Packet) {
if (null != packet.id) { if (null != packet.id) {
// debug('attaching ack callback to event'); // debug('attaching ack callback to event');
this.dispatch(packet, this.ack(packet.id)); this.dispatch(packet, this.ack(packet.id))
} else { } else {
this.dispatch(packet); this.dispatch(packet)
} }
} }
ack(id: number) { ack(id: number) {
let sent = false; let sent = false
return (...args: any[]) => { return (...args: any[]) => {
if (sent) return; if (sent) return
this.packet({ this.packet({
id: id, id: id,
type: PacketTypes.MESSAGE, type: PacketTypes.MESSAGE,
sub_type: this.hasBin(args) ? SubPacketTypes.BINARY_ACK : SubPacketTypes.ACK, sub_type: this.hasBin(args) ? SubPacketTypes.BINARY_ACK : SubPacketTypes.ACK,
data: args data: args
}); })
sent = true; sent = true
} }
} }
onack(packet: Packet) { onack(packet: Packet) {
let ack = this.acks[packet.id]; let ack = this.acks[packet.id]
if ('function' == typeof ack) { if ('function' == typeof ack) {
// debug('calling ack %s with %j', packet.id, packet.data); // debug('calling ack %s with %j', packet.id, packet.data);
ack.apply(this, packet.data); ack.apply(this, packet.data)
delete this.acks[packet.id]; delete this.acks[packet.id]
} else { } else {
// debug('bad ack %s', packet.id); // debug('bad ack %s', packet.id);
} }
} }
dispatch(packet: Packet, ack?: Function) { dispatch(packet: Packet, ack?: Function) {
if (ack) { this.acks[packet.id] = ack; } if (ack) { this.acks[packet.id] = ack }
super.emit(packet.name, ...packet.data, ack) super.emit(packet.name, ...packet.data, ack)
} }
private hasBin(obj: any) { private hasBin(obj: any) {
return false; return false
} }
} }

51
packages/websocket/src/tomcat/client.ts
View File

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