diff --git a/packages/net/src/Connection.ts b/packages/net/src/Connection.ts index 1d4c9bf..041e750 100644 --- a/packages/net/src/Connection.ts +++ b/packages/net/src/Connection.ts @@ -1,4 +1,4 @@ -import {Emitter, Worker, sleep} from '@welshman/lib' +import {ctx, Emitter, Worker, sleep} from '@welshman/lib' import {AuthStatus, ConnectionMeta} from './ConnectionMeta' import {Socket, isMessage, asMessage} from './Socket' import type {SocketMessage, Message} from './Socket' @@ -93,6 +93,20 @@ export class Connection extends Emitter { } onReceive = (message: SocketMessage) => { + const [verb, ...extra] = asMessage(message) + + if (verb === 'AUTH') { + const [challenge] = extra + + ctx.net.onAuth(this.url, challenge) + } + + if (verb === 'OK') { + const [id, ok, message] = extra + + ctx.net.onOk(this.url, id, ok, message) + } + this.emit('receive', this, message) } @@ -120,11 +134,14 @@ export class Connection extends Emitter { await Promise.race([ sleep(timeout), new Promise(resolve => { - this.on('receive', (cxn: Connection, message: Message) => { + const onReceive = (cxn: Connection, message: Message) => { if (message[0] === 'OK' && message[2]) { + this.off('receive', onReceive) resolve() } - }) + } + + this.on('receive', onReceive) }) ]) } diff --git a/packages/net/src/Executor.ts b/packages/net/src/Executor.ts index 83758c8..0a6ad41 100644 --- a/packages/net/src/Executor.ts +++ b/packages/net/src/Executor.ts @@ -30,10 +30,7 @@ const createSubId = (prefix: string) => [prefix, Math.random().toString().slice( export class Executor { - constructor(readonly target: Target) { - target.on('AUTH', ctx.net.onAuth) - target.on('OK', ctx.net.onOk) - } + constructor(readonly target: Target) {} subscribe(filters: Filter[], {onEvent, onEose}: SubscribeOpts = {}) { let closed = false