Refactor everything to use EventEmitter

.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"],
     }
 }

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"

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({
+    const eventListener = (url, subid, e) => subid === id && onEvent?.(url, e)
-      EVENT: (url, subid, e) => subid === id && onEvent?.(url, e),
+    const eoseListener = (url, subid) => subid === id && onEose?.(url)
-      EOSE: (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)
-
+        this.target.off('EVENT', eventListener)
-        unsubscribe()
+        this.target.off('EOSE', eoseListener)
       },
     }
   }
   publish(event, {verb = 'EVENT', onOk, onError}) {
-    const unsubscribe = this.target.bus.addListeners({
+    const okListener = (url, id, ...payload) => id === event.id && onOk(url, id, ...payload)
-      OK: (url, id, ...payload) => id === event.id && onOk(url, id, ...payload),
+    const errorListener = (url, id, ...payload) => id === event.id && onError(url, id, ...payload)
-      ERROR: (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({
+    const countListener = (url, subid, ...payload) => {
-      COUNT: (url, subid, ...payload) => {
+      if (subid === id) {
-        if (subid === id) {
+        onCount(url, ...payload)
-          onCount(url, ...payload)
+        this.target.off('COUNT', countListener)
-          unsubscribe()
-        }
       }
-    })
+    }
 
+    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)
+      }
+    }
   }
 }

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.socket.on('message', this.onMessage)
-    this.unsubscribe = socket.bus.addListeners({
-      message: (websocketUrl, [{relays}, [verb, ...payload]]) => {
-        this.bus.emit(verb, relays[0], ...payload)
-      },
-    })
   }
   get sockets() {
     return [this.socket]
   }
-  async send(...payload) {
+  send = (...payload) => {
-    await this.socket.connect()
-
     this.socket.send([{relays: this.urls}, payload])
   }
-  cleanup() {
+  onMessage = (websocketUrl, [{relays}, [verb, ...payload]]) => {
-    this.bus.clear()
+    this.emit(verb, relays[0], ...payload)
-    this.unsubscribe()
+  }
+  cleanup = () => {
+    this.socket.off('message', this.onMessage)
   }
 }

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<string, Socket>
   constructor() {
+    super()
+
     this.data = new Map()
-    this.bus = new EventBus()
   }
   has(url) {
     return this.data.has(url)
   }
-  get(url) {
+  get(url, {autoConnect = true} = {}) {
-    if (!this.data.has(url)) {
+    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({
+      socket.on('open', () => this.emit('open', {url}))
-        open: () => this.bus.emit('open', {url}),
+      socket.on('close', () => this.emit('close', {url}))
-        close: () => this.bus.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)
     }
   }

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.socket.on('message', this.onMessage)
-    this.listeners = [
-      socket.bus.addListener('message', (url, [verb, ...payload]) => {
-        this.bus.emit(verb, url, ...payload)
-      })
-    ]
   }
   get sockets() {
     return [this.socket]
   }
-  async send(...payload) {
+  send(...payload) {
-    await this.socket.connect()
-
     this.socket.send(payload)
   }
-  cleanup() {
+  onMessage = (url, [verb, ...payload]) => {
-    this.bus.clear()
+    this.emit(verb, url, ...payload)
-    this.listeners.map(unsubscribe => unsubscribe())
+  }
+  cleanup = () => {
+    this.socket.off('message', this.onMessage)
   }
 }

src/Relays.ts

@@ -1,27 +1,25 @@
-import {Socket} from './util/Socket'
+import {EventEmitter} from 'events'
-import {EventBus} from './util/EventBus'
 
-export class Relays {
+export class Relays extends EventEmitter {
-  sockets: Socket[]
-  bus: EventBus
   constructor(sockets) {
+    super()
+
     this.sockets = sockets
-    this.bus = new EventBus()
+    this.sockets.forEach(socket => {
-    this.listeners = sockets.map(socket => {
+      socket.on('message', this.onMessage)
-      return socket.bus.addListener('message', (url, [verb, ...payload]) => {
-        this.bus.emit(verb, url, ...payload)
-      })
     })
   }
-  send(...payload) {
+  send = (...payload) => {
-    this.sockets.forEach(async socket => {
+    this.sockets.forEach(socket => {
-      await socket.connect()
-
       socket.send(payload)
     })
   }
-  cleanup() {
+  onMessage = (url, [verb, ...payload]) => {
-    this.bus.clear()
+    this.emit(verb, url, ...payload)
-    this.listeners.map(unsubscribe => unsubscribe())
+  }
+  cleanup = () => {
+    this.sockets.forEach(socket => {
+      socket.off('message', this.onMessage)
+    })
   }
 }

src/main.ts

@@ -1,4 +1,3 @@
-export * from "./util/EventBus"
 export * from "./util/Deferred"
 export * from "./util/Socket"
 export * from "./Executor"

src/util/EventBus.ts

@@ -1,35 +0,0 @@
-export type EventBusHandler = (...args: any[]) => void
-
-export class EventBus {
-  static ANY = Math.random().toString().slice(2)
-  listeners: Record<string, Array<EventBusHandler>> = {}
-  addListener(name: string, handler: EventBusHandler) {
-    this.listeners[name] = this.listeners[name] || ([] as Array<EventBusHandler>)
-    this.listeners[name].push(handler)
-
-    return () => this.removeListener(name, handler)
-  }
-  addListeners(config: Record<string, EventBusHandler>) {
-    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)
-    }
-  }
-}

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<void>
   timeout?: NodeJS.Timeout
-  queue: string[]
+  queue: [string, any][]
-  bus: EventBus
   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) {
+  onError = (error: Error) => {
-    this.queue.push(event.data as string)
-
-    if (!this.timeout) {
-      this.handleMessagesAsync()
-    }
-  }
-  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() {
+  connect = () => {
-    if ([Socket.STATUS.NEW, Socket.STATUS.CLOSED].includes(this.status)) {
+    const {NEW, CLOSED, PENDING} = Socket.STATUS
-      if (this.ws) {
-        console.error("Attempted to connect when already connected", this)
-      }
 
+    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("close", this.onClose)
-      this.ws.addEventListener("open", this._onOpen)
+      // @ts-ignore
-      this.ws.addEventListener("message", this._onMessage)
+      this.ws.addEventListener("error", this.onError)
-      this.ws.addEventListener("error", this._onError)
+      // @ts-ignore
-      this.ws.addEventListener("close", this._onClose)
+      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() {
+  receiveMessage = (json: string) => {
-    this.disconnect()
+    try {
-    this.bus.clear()
+      this.emit('message', this.url, JSON.parse(json))
-  }
+    } catch (e) {
-  handleMessages() {
+      // pass
-    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
     }
   }
-  handleMessagesAsync() {
+  sendMessage = (message: any) => {
-    this.timeout = setTimeout(() => this.handleMessages(), 10) as NodeJS.Timeout
+    // @ts-ignore
+    this.ws.send(JSON.stringify(message))
   }
-  send(message: any) {
+  shouldDeferWork = () => {
-    if (this.status === Socket.STATUS.READY) {
+    // These sometimes get out of sync
-      if (this.ws?.readyState !== 1) {
+    return this.status !== Socket.STATUS.READY || this.ws?.readyState !== 1
-        console.warn("Send attempted before socket was ready", this)
+  }
+  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
     }
   }
 }

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"