coracle/welshman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve consistency of interface

Browse Source
This commit is contained in:
Jonathan Staab
2023-03-27 15:05:34 -05:00
parent 853b42c1c9
commit 9b6a779397
9 changed files with 223 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/Executor.ts
+2 -2
View File
@@ -1,12 +1,12 @@
import type {EventBus} from './util/EventBus.ts'
type ExecutorTarget = {
type Executable = {
bus: EventBus
send: (verb: string, ...args) => void
}
export class Executor {
target: ExecutorTarget
target: Executable
constructor(target) {
this.target = target
}
lib/Plex.ts
+22
View File
@@ -0,0 +1,22 @@
import {EventBus} from "./util/EventBus"
export class Plex {
constructor(urls, socket) {
this.urls = urls
this.socket = socket
this.bus = new EventBus()
this.onMessage = this.onMessage.bind(this)
this.socket.bus.on('message', this.onMessage)
}
async send(...payload) {
await this.socket.connect()
this.socket.send([{relays: this.urls}, payload])
}
onMessage(message) {
const [verb, ...payload] = message[1]
this.bus.handle(verb, ...payload)
}
}
lib/Pool.ts
+3 -16
View File
@@ -1,9 +1,7 @@
import {Relay} from "./Relay"
const normalizeUrl = url => url.replace(/\/+$/, "").toLowerCase().trim()
import {Socket} from "./util/Socket"
export class Pool {
relays: Map<string, Relay>
relays: Map<string, Socket>
constructor() {
this.relays = new Map()
this.interval = setInterval(() => {
@@ -13,17 +11,13 @@ export class Pool {
}, 30_000)
}
add(url) {
url = normalizeUrl(url)
if (!this.relays.has(url)) {
this.relays.set(url, new Relay(url))
this.relays.set(url, new Socket(url))
}
return this.relays.get(url)
}
remove(url) {
url = normalizeUrl(url)
this.relays.get(url)?.disconnect()
this.relays.delete(url)
}
@@ -34,11 +28,4 @@ export class Pool {
this.remove(url)
}
}
async waitFor(url) {
const relay = this.add(url)
await relay.connect()
return relay.status === Relay.STATUS.READY ? relay : null
}
}
lib/Relay.ts
+11 -108
View File
@@ -1,118 +1,21 @@
import WebSocket from "isomorphic-ws"
import {EventBus} from "./util/EventBus"
import {Deferred, defer} from "./util/Deferred"
export class Relay {
ws?: WebSocket
url: string
ready?: Deferred<void>
queue: string[]
error: string
status: string
timeout?: NodeJS.Timeout
bus: EventBus
static STATUS = {
NEW: "new",
PENDING: "pending",
CLOSED: "closed",
ERROR: "error",
READY: "ready",
}
static ERROR = {
CONNECTION: "connection",
UNAUTHORIZED: "unauthorized",
FORBIDDEN: "forbidden",
}
constructor(url) {
this.ws = null
this.url = url
this.ready = null
this.queue = []
this.timeout = null
constructor(socket) {
this.socket = socket
this.bus = new EventBus()
this.error = null
this.status = Relay.STATUS.NEW
this.onMessage = this.onMessage.bind(this)
this.socket.bus.on('message', this.onMessage)
}
async connect() {
if (this.status === Relay.STATUS.NEW) {
if (this.ws) {
console.error("Attempted to connect when already connected", this)
}
async send(...payload) {
await this.socket.connect()
this.ready = defer()
this.ws = new WebSocket(this.url)
this.status = Relay.STATUS.PENDING
this.ws.addEventListener("open", () => {
console.log(`Opened connection to ${this.url}`)
this.status = Relay.STATUS.READY
this.ready.resolve()
})
this.ws.addEventListener("message", e => {
this.queue.push(e.data)
if (!this.timeout) {
this.timeout = setTimeout(() => this.handleMessages(), 10)
}
})
this.ws.addEventListener("error", e => {
console.log(`Error on connection to ${this.url}`)
this.disconnect()
this.ready.reject()
this.error = Relay.ERROR.CONNECTION
this.status = Relay.STATUS.CLOSED
})
this.ws.addEventListener("close", () => {
console.log(`Closed connection to ${this.url}`)
this.disconnect()
this.ready.reject()
this.status = Relay.STATUS.CLOSED
})
}
await this.ready.catch(() => null)
this.socket.send(payload)
}
reconnect() {
if (this.status === Relay.STATUS.ERROR) {
this.status = Relay.STATUS.NEW
this.connect()
}
}
disconnect() {
if (this.ws) {
console.log(`Disconnecting from ${this.url}`)
onMessage(message) {
const [verb, ...payload] = message
this.ws.close()
this.ws = null
}
}
handleMessages() {
for (const json of this.queue.splice(0, 10)) {
let message
try {
message = JSON.parse(json)
} catch (e) {
continue
}
const [verb, ...args] = message
this.bus.handle(verb, ...args)
}
this.timeout = this.queue.length > 0 ? setTimeout(() => this.handleMessages(), 10) : null
}
send(...payload) {
if (this.ws?.readyState !== 1) {
console.warn("Send attempted before socket was ready", this)
}
this.ws.send(JSON.stringify(payload))
this.bus.handle(verb, ...payload)
}
}
lib/RelaySet.ts
-24
View File
@@ -1,24 +0,0 @@
import type {Relay} from './Relay'
import {EventBus} from './util/EventBus'
export class RelaySet {
relays: Relay[]
bus: EventBus
constructor(relays) {
this.relays = relays
this.bus = new EventBus()
relays.forEach(relay => {
relay.bus.pipe(EventBus.ANY, this.bus)
})
}
send(...payload) {
this.relays.forEach(async relay => {
await relay.connect()
if (relay.status === Relay.STATUS.READY) {
relay.send(...payload)
}
})
}
}
lib/Relays.ts
+26
View File
@@ -0,0 +1,26 @@
import {Socket} from './util/Socket'
import {EventBus} from './util/EventBus'
export class Relays {
sockets: Socket[]
bus: EventBus
constructor(sockets) {
this.sockets = sockets
this.bus = new EventBus()
this.onMessage = this.onMessage.bind(this)
sockets.forEach(socket => socket.bus.on('message', this.onMessage))
}
send(...payload) {
this.sockets.forEach(socket => {
await socket.connect()
socket.send(...payload)
})
}
onMessage(message) {
const [verb, ...payload] = message
this.bus.handle(verb, ...payload)
}
}
lib/util/EventBus.ts
+7 -15
View File
@@ -1,32 +1,24 @@
export type EventBusHandler = (...args: any[]) => void
export type EventBusListener = {
id: string
handler: EventBusHandler
}
export class EventBus {
static ANY = Math.random().toString().slice(2)
listeners: Record<string, Array<EventBusListener>> = {}
listeners: Record<string, Array<EventBusHandler>> = {}
on(name: string, handler: EventBusHandler) {
const id = Math.random().toString().slice(2)
this.listeners[name] = this.listeners[name] || ([] as Array<EventBusListener>)
this.listeners[name].push({id, handler})
return id
this.listeners[name] = this.listeners[name] || ([] as Array<EventBusHandler>)
this.listeners[name].push(handler)
}
off(name: string, id: string) {
this.listeners[name] = this.listeners[name].filter(l => l.id !== id)
off(name: string, handler: EventBusHandler) {
this.listeners[name] = this.listeners[name].filter(h => h !== handler)
}
clear() {
this.listeners = {}
}
handle(k: string, ...payload: any) {
for (const {handler} of this.listeners[k] || []) {
for (const handler of this.listeners[k] || []) {
handler(...payload)
}
for (const {handler} of this.listeners[EventBus.ANY] || []) {
for (const handler of this.listeners[EventBus.ANY] || []) {
handler(k, ...payload)
}
}
lib/util/Socket.ts
+106
View File
@@ -0,0 +1,106 @@
import WebSocket from "isomorphic-ws"
import {EventBus} from "./EventBus"
import {Deferred, defer} from "./Deferred"
export class Socket {
ws?: WebSocket
url: string
ready?: Deferred<void>
timeout?: NodeJS.Timeout
queue: string[]
bus: EventBus
status: string
static STATUS = {
NEW: "new",
PENDING: "pending",
CLOSED: "closed",
READY: "ready",
}
constructor(url: string) {
this.ws = undefined
this.url = url
this.ready = undefined
this.timeout = undefined
this.queue = []
this.bus = new EventBus()
this.status = Socket.STATUS.NEW
}
async connect() {
if ([Socket.STATUS.NEW, Socket.STATUS.CLOSED].includes(this.status)) {
if (this.ws) {
console.error("Attempted to connect when already connected", this)
}
this.ready = defer()
this.ws = new WebSocket(this.url)
this.status = Socket.STATUS.PENDING
this.ws.addEventListener("open", () => {
console.log(`Opened connection to ${this.url}`)
this.status = Socket.STATUS.READY
this.ready?.resolve()
})
this.ws.addEventListener("message", e => {
this.queue.push(e.data as string)
if (!this.timeout) {
this.timeout = this.handleMessagesAsync()
}
})
this.ws.addEventListener("error", e => {
console.log(`Error on connection to ${this.url}`)
this.disconnect()
this.ready?.reject()
this.status = Socket.STATUS.CLOSED
})
this.ws.addEventListener("close", () => {
console.log(`Closed connection to ${this.url}`)
this.disconnect()
this.ready?.reject()
this.status = Socket.STATUS.CLOSED
})
}
await this.ready?.catch(() => null)
}
disconnect() {
if (this.ws) {
console.log(`Disconnecting from ${this.url}`)
this.ws.close()
this.ws = undefined
}
}
handleMessages() {
for (const json of this.queue.splice(0, 10)) {
let message
try {
message = JSON.parse(json)
} catch (e) {
continue
}
this.bus.handle('message', message)
}
this.timeout = this.queue.length > 0 ? this.handleMessagesAsync() : undefined
}
handleMessagesAsync() {
return setTimeout(() => this.handleMessages(), 10) as NodeJS.Timeout
}
send(message: any) {
if (this.status === Socket.STATUS.READY) {
if (this.ws?.readyState !== 1) {
console.warn("Send attempted before socket was ready", this)
}
this.ws?.send(JSON.stringify(message))
}
}
}