From 4d6ea603ff59e0285f4049ae9eab9d29309d3303 Mon Sep 17 00:00:00 2001 From: Jonathan Staab Date: Fri, 7 Jul 2023 17:37:35 -0700 Subject: [PATCH] Refactor everything to use EventEmitter --- .eslintrc.js | 3 + package.json | 3 +- src/Executor.ts | 63 +++++++++++--------- src/Plex.ts | 25 ++++---- src/Pool.ts | 21 ++++--- src/Relay.ts | 25 ++++---- src/Relays.ts | 32 +++++------ src/main.ts | 1 - src/util/EventBus.ts | 35 ------------ src/util/Socket.ts | 133 +++++++++++++++++++++---------------------- yarn.lock | 5 ++ 11 files changed, 157 insertions(+), 189 deletions(-) delete mode 100644 src/util/EventBus.ts diff --git a/.eslintrc.js b/.eslintrc.js index 8ed72c1..decfa28 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -20,5 +20,8 @@ module.exports = { "@typescript-eslint/no-unused-vars": ["error", {args: "none"}], "@typescript-eslint/no-explicit-any": "off", "@typescript-eslint/ban-ts-comment": "off", + "object-curly-spacing": ["error", "never"], + "array-bracket-spacing": ["error", "never"], + "semi": ["error", "never"], } } diff --git a/package.json b/package.json index 21afd9c..663c794 100644 --- a/package.json +++ b/package.json @@ -16,7 +16,7 @@ "build": "node build.js", "pub": "npm i && npm run lint && node build.js && npm publish", "lint:ts": "tsc --noEmit --esModuleInterop --strict src/**/*", - "lint:es": "eslint src/*", + "lint:es": "eslint src/* --fix", "lint": "run-p lint:*" }, "keywords": [ @@ -44,6 +44,7 @@ "typescript": "^4.9.4" }, "dependencies": { + "events": "^3.3.0", "husky": "^8.0.3", "isomorphic-ws": "^5.0.0", "npm-run-all": "^4.1.5" diff --git a/src/Executor.ts b/src/Executor.ts index d50027d..78e5e85 100644 --- a/src/Executor.ts +++ b/src/Executor.ts @@ -1,62 +1,69 @@ -import type {EventBus} from './util/EventBus.ts' +import {EventEmitter} from 'events' const createSubId = prefix => [prefix, Math.random().toString().slice(2, 10)].join('-') -type Executable = { - bus: EventBus - send: (verb: string, ...args) => void -} - export class Executor { - target: Executable + target: EventEmitter constructor(target) { this.target = target } subscribe(filters, {onEvent, onEose}) { const id = createSubId('REQ') - const unsubscribe = this.target.bus.addListeners({ - EVENT: (url, subid, e) => subid === id && onEvent?.(url, e), - EOSE: (url, subid) => subid === id && onEose?.(url), - }) + const eventListener = (url, subid, e) => subid === id && onEvent?.(url, e) + const eoseListener = (url, subid) => subid === id && onEose?.(url) + this.target.on('EVENT', eventListener) + this.target.on('EOSE', eoseListener) this.target.send("REQ", id, ...filters) return { unsubscribe: () => { this.target.send("CLOSE", id) - - unsubscribe() + this.target.off('EVENT', eventListener) + this.target.off('EOSE', eoseListener) }, } } publish(event, {verb = 'EVENT', onOk, onError}) { - const unsubscribe = this.target.bus.addListeners({ - OK: (url, id, ...payload) => id === event.id && onOk(url, id, ...payload), - ERROR: (url, id, ...payload) => id === event.id && onError(url, id, ...payload), - }) + const okListener = (url, id, ...payload) => id === event.id && onOk(url, id, ...payload) + const errorListener = (url, id, ...payload) => id === event.id && onError(url, id, ...payload) + this.target.on('OK', okListener) + this.target.on('ERROR', errorListener) this.target.send(verb, event) - return {unsubscribe} + return { + unsubscribe: () => { + this.target.off('OK', okListener) + this.target.off('ERROR', errorListener) + } + } } count(filters, {onCount}) { const id = createSubId('COUNT') - const unsubscribe = this.target.bus.addListeners({ - COUNT: (url, subid, ...payload) => { - if (subid === id) { - onCount(url, ...payload) - unsubscribe() - } + const countListener = (url, subid, ...payload) => { + if (subid === id) { + onCount(url, ...payload) + this.target.off('COUNT', countListener) } - }) + } + this.target.on('COUNT', countListener) this.target.send("COUNT", id, ...filters) - return {unsubscribe} + return { + unsubscribe: () => this.target.off('COUNT', countListener) + } } handleAuth({onAuth, onOk}) { - const unsubscribe = this.target.bus.addListeners({AUTH: onAuth, OK: onOk}) + this.target.on('AUTH', onAuth) + this.target.on('OK', onOk) - return {unsubscribe} + return { + unsubscribe: () => { + this.target.off('AUTH', onAuth) + this.target.off('OK', onOk) + } + } } } diff --git a/src/Plex.ts b/src/Plex.ts index 0f612d4..954f6c9 100644 --- a/src/Plex.ts +++ b/src/Plex.ts @@ -1,26 +1,23 @@ -import {EventBus} from "./util/EventBus" +import {EventEmitter} from 'events' -export class Plex { +export class Plex extends EventEmitter { constructor(urls, socket) { + super() + this.urls = urls this.socket = socket - this.bus = new EventBus() - this.unsubscribe = socket.bus.addListeners({ - message: (websocketUrl, [{relays}, [verb, ...payload]]) => { - this.bus.emit(verb, relays[0], ...payload) - }, - }) + this.socket.on('message', this.onMessage) } get sockets() { return [this.socket] } - async send(...payload) { - await this.socket.connect() - + send = (...payload) => { this.socket.send([{relays: this.urls}, payload]) } - cleanup() { - this.bus.clear() - this.unsubscribe() + onMessage = (websocketUrl, [{relays}, [verb, ...payload]]) => { + this.emit(verb, relays[0], ...payload) + } + cleanup = () => { + this.socket.off('message', this.onMessage) } } diff --git a/src/Pool.ts b/src/Pool.ts index 9266ce8..3f3cc70 100644 --- a/src/Pool.ts +++ b/src/Pool.ts @@ -1,26 +1,25 @@ import {Socket} from "./util/Socket" -import {EventBus} from "./util/EventBus" +import {EventEmitter} from 'events' -export class Pool { +export class Pool extends EventEmitter { data: Map constructor() { + super() + this.data = new Map() - this.bus = new EventBus() } has(url) { return this.data.has(url) } - get(url) { - if (!this.data.has(url)) { + get(url, {autoConnect = true} = {}) { + if (!this.data.has(url) && autoConnect) { const socket = new Socket(url) this.data.set(url, socket) - this.bus.emit('init', {url}) + this.emit('init', {url}) - socket.bus.addListeners({ - open: () => this.bus.emit('open', {url}), - close: () => this.bus.emit('close', {url}), - }) + socket.on('open', () => this.emit('open', {url})) + socket.on('close', () => this.emit('close', {url})) } return this.data.get(url) @@ -29,7 +28,7 @@ export class Pool { const socket = this.data.get(url) if (socket) { - socket.cleanup() + socket.removeAllListeners() this.data.delete(url) } } diff --git a/src/Relay.ts b/src/Relay.ts index 8792f27..a77e48e 100644 --- a/src/Relay.ts +++ b/src/Relay.ts @@ -1,25 +1,22 @@ -import {EventBus} from "./util/EventBus" +import {EventEmitter} from 'events' -export class Relay { +export class Relay extends EventEmitter { constructor(socket) { + super() + this.socket = socket - this.bus = new EventBus() - this.listeners = [ - socket.bus.addListener('message', (url, [verb, ...payload]) => { - this.bus.emit(verb, url, ...payload) - }) - ] + this.socket.on('message', this.onMessage) } get sockets() { return [this.socket] } - async send(...payload) { - await this.socket.connect() - + send(...payload) { this.socket.send(payload) } - cleanup() { - this.bus.clear() - this.listeners.map(unsubscribe => unsubscribe()) + onMessage = (url, [verb, ...payload]) => { + this.emit(verb, url, ...payload) + } + cleanup = () => { + this.socket.off('message', this.onMessage) } } diff --git a/src/Relays.ts b/src/Relays.ts index 23a93e9..c9a8cd1 100644 --- a/src/Relays.ts +++ b/src/Relays.ts @@ -1,27 +1,25 @@ -import {Socket} from './util/Socket' -import {EventBus} from './util/EventBus' +import {EventEmitter} from 'events' -export class Relays { - sockets: Socket[] - bus: EventBus +export class Relays extends EventEmitter { constructor(sockets) { + super() + this.sockets = sockets - this.bus = new EventBus() - this.listeners = sockets.map(socket => { - return socket.bus.addListener('message', (url, [verb, ...payload]) => { - this.bus.emit(verb, url, ...payload) - }) + this.sockets.forEach(socket => { + socket.on('message', this.onMessage) }) } - send(...payload) { - this.sockets.forEach(async socket => { - await socket.connect() - + send = (...payload) => { + this.sockets.forEach(socket => { socket.send(payload) }) } - cleanup() { - this.bus.clear() - this.listeners.map(unsubscribe => unsubscribe()) + onMessage = (url, [verb, ...payload]) => { + this.emit(verb, url, ...payload) + } + cleanup = () => { + this.sockets.forEach(socket => { + socket.off('message', this.onMessage) + }) } } diff --git a/src/main.ts b/src/main.ts index 791ef39..78878e5 100644 --- a/src/main.ts +++ b/src/main.ts @@ -1,4 +1,3 @@ -export * from "./util/EventBus" export * from "./util/Deferred" export * from "./util/Socket" export * from "./Executor" diff --git a/src/util/EventBus.ts b/src/util/EventBus.ts deleted file mode 100644 index c2745ea..0000000 --- a/src/util/EventBus.ts +++ /dev/null @@ -1,35 +0,0 @@ -export type EventBusHandler = (...args: any[]) => void - -export class EventBus { - static ANY = Math.random().toString().slice(2) - listeners: Record> = {} - addListener(name: string, handler: EventBusHandler) { - this.listeners[name] = this.listeners[name] || ([] as Array) - this.listeners[name].push(handler) - - return () => this.removeListener(name, handler) - } - addListeners(config: Record) { - const callbacks = [] as Array<() => void> - for (const [name, handler] of Object.entries(config)) { - callbacks.push(this.addListener(name, handler)) - } - - return () => callbacks.forEach(unsubscribe => unsubscribe()) - } - removeListener(name: string, handler: EventBusHandler) { - this.listeners[name] = (this.listeners[name] || []).filter(h => h !== handler) - } - clear() { - this.listeners = {} - } - emit(k: string, ...payload: any) { - for (const handler of this.listeners[k] || []) { - handler(...payload) - } - - for (const handler of this.listeners[EventBus.ANY] || []) { - handler(k, ...payload) - } - } -} diff --git a/src/util/Socket.ts b/src/util/Socket.ts index 91cc29f..2c5161f 100644 --- a/src/util/Socket.ts +++ b/src/util/Socket.ts @@ -1,20 +1,15 @@ import WebSocket from "isomorphic-ws" -import {EventBus} from "./EventBus" +import {EventEmitter} from 'events' import {Deferred, defer} from "./Deferred" -export class Socket { +export class Socket extends EventEmitter { ws?: WebSocket url: string ready: Deferred timeout?: NodeJS.Timeout - queue: string[] - bus: EventBus + queue: [string, any][] status: string error?: Error - _onOpen: (e: any) => void - _onMessage: (e: any) => void - _onError: (e: any) => void - _onClose: (e: any) => void static STATUS = { NEW: "new", PENDING: "pending", @@ -22,99 +17,101 @@ export class Socket { READY: "ready", } constructor(url: string) { + super() + this.url = url this.ready = defer() this.queue = [] - this.bus = new EventBus() this.status = Socket.STATUS.NEW + + this.setMaxListeners(100) } - onOpen() { + send = (message: any) => { + this.connect() + this.queue.push(['send', message]) + this.enqueueWork() + } + onMessage = (event: {data: string}) => { + this.queue.push(['receive', event.data]) + this.enqueueWork() + } + onOpen = () => { this.error = undefined this.status = Socket.STATUS.READY this.ready.resolve() - this.bus.emit('open') + this.emit('open') } - onMessage(event) { - this.queue.push(event.data as string) - - if (!this.timeout) { - this.handleMessagesAsync() - } - } - onError(error: Error) { + onError = (error: Error) => { this.error = error - this.bus.emit('error', error) + this.emit('fault', error) } - onClose() { + onClose = () => { this.disconnect() this.ready.reject() this.status = Socket.STATUS.CLOSED - this.bus.emit('close') + this.emit('close') } - async connect() { - if ([Socket.STATUS.NEW, Socket.STATUS.CLOSED].includes(this.status)) { - if (this.ws) { - console.error("Attempted to connect when already connected", this) - } + connect = () => { + const {NEW, CLOSED, PENDING} = Socket.STATUS + if ([NEW, CLOSED].includes(this.status)) { this.ready = defer() + this.status = PENDING this.ws = new WebSocket(this.url) - this.status = Socket.STATUS.PENDING - - this.ws.addEventListener("open", this._onOpen) - this.ws.addEventListener("message", this._onMessage) - this.ws.addEventListener("error", this._onError) - this.ws.addEventListener("close", this._onClose) + this.ws.addEventListener("open", this.onOpen) + this.ws.addEventListener("close", this.onClose) + // @ts-ignore + this.ws.addEventListener("error", this.onError) + // @ts-ignore + this.ws.addEventListener("message", this.onMessage) } - - await this.ready.catch(() => null) } - disconnect() { + disconnect = () => { if (this.ws) { const ws = this.ws // Avoid "WebSocket was closed before the connection was established" this.ready.then(() => ws.close(), () => null) - - this.ws.removeEventListener("open", this._onOpen) - this.ws.removeEventListener("message", this._onMessage) - this.ws.removeEventListener("error", this._onError) - this.ws.removeEventListener("close", this._onClose) this.ws = undefined } } - cleanup() { - this.disconnect() - this.bus.clear() - } - handleMessages() { - for (const json of this.queue.splice(0, 10)) { - let message - try { - message = JSON.parse(json) - } catch (e) { - continue - } - - this.bus.emit('message', this.url, message) - } - - if (this.queue.length > 0) { - this.handleMessagesAsync() - } else { - this.timeout = undefined + receiveMessage = (json: string) => { + try { + this.emit('message', this.url, JSON.parse(json)) + } catch (e) { + // pass } } - handleMessagesAsync() { - this.timeout = setTimeout(() => this.handleMessages(), 10) as NodeJS.Timeout + sendMessage = (message: any) => { + // @ts-ignore + this.ws.send(JSON.stringify(message)) } - send(message: any) { - if (this.status === Socket.STATUS.READY) { - if (this.ws?.readyState !== 1) { - console.warn("Send attempted before socket was ready", this) + shouldDeferWork = () => { + // These sometimes get out of sync + return this.status !== Socket.STATUS.READY || this.ws?.readyState !== 1 + } + doWork = () => { + this.timeout = undefined + + for (const [action, payload] of this.queue.splice(0, 50)) { + if (action === 'receive') { + this.receiveMessage(payload) } - this.ws?.send(JSON.stringify(message)) + if (action === 'send') { + if (this.shouldDeferWork()) { + this.queue.push(['send', payload]) + } else { + this.sendMessage(payload) + } + } + } + + this.enqueueWork() + } + enqueueWork = () => { + if (!this.timeout && this.queue.length > 0) { + this.timeout = setTimeout(() => this.doWork(), 50) as NodeJS.Timeout } } } diff --git a/yarn.lock b/yarn.lock index 3989f52..7f9a9c3 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1229,6 +1229,11 @@ esutils@^2.0.2: resolved "https://registry.yarnpkg.com/esutils/-/esutils-2.0.3.tgz#74d2eb4de0b8da1293711910d50775b9b710ef64" integrity sha512-kVscqXk4OCp68SZ0dkgEKVi6/8ij300KBWTJq32P/dYeWTSwK41WyTxalN1eRmA5Z9UU/LX9D7FWSmV9SAYx6g== +events@^3.3.0: + version "3.3.0" + resolved "https://registry.yarnpkg.com/events/-/events-3.3.0.tgz#31a95ad0a924e2d2c419a813aeb2c4e878ea7400" + integrity sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q== + fast-deep-equal@^3.1.1, fast-deep-equal@^3.1.3: version "3.1.3" resolved "https://registry.yarnpkg.com/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz#3a7d56b559d6cbc3eb512325244e619a65c6c525"