laantungir/nostr-tools
1
0
Fork 0
mirror of https://github.com/nbd-wtf/nostr-tools.git synced 2025-12-09 16:48:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #41 from monlovesmango/cb-api

Browse Source 
refactor of cb api
This commit is contained in:
fiatjaf
2022-12-17 13:06:30 -03:00
committed by GitHub
parent 12be5a5338 100c77d2aa
commit e62b833464
4 changed files with 246 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
README.md
View File

@@ -14,41 +14,59 @@ pool.setPrivateKey('<hex>') // optional
pool.addRelay('ws://some.relay.com', {read: true, write: true}) pool.addRelay('ws://some.relay.com', {read: true, write: true})
pool.addRelay('ws://other.relay.cool', {read: true, write: true}) pool.addRelay('ws://other.relay.cool', {read: true, write: true})
// example callback function for a subscription // example callback functions for listeners
function onEvent(event, relay) { // callback functions take an object argument with following keys:
console.log(`got an event from ${relay.url} which is already validated.`, event) // - relay: relay url
// - type: type of listener
// - id: sub id for sub specific listeners ('EVENT' or 'EOSE')
// - event: event object, only for 'event' listener
// - notice: notice message, only for 'notice' listener
function onEvent({event, relay, type, id}) {
console.log(`got an event from ${relay} which is already validated.`, event)
} }
function onEose({relay, type, id}) { /* callback function here */}
function onNotice({relay, type, notice}) { /* callback function here */}
function onConnection({relay, type}) { /* callback function here */}
// listen for messages for pool
pool.on('event', onEvent)
pool.on('connection', onConnection)
pool.on('notice', onNotice)
// subscribing to a single user // subscribing to a single user
// author is the user's public key // author is the user's public key
pool.sub({cb: onEvent, filter: {author: '<hex>'}}) pool.sub({filter: {author: '<hex>'}})
// or bulk follow // or bulk follow
pool.sub({cb:(event, relay) => {...}, filter: {authors: ['<hex1>', '<hex2>', ..., '<hexn>']}}) pool.sub({filter: {authors: ['<hex1>', '<hex2>', ..., '<hexn>']}})
// reuse a subscription channel // reuse a subscription channel
const mySubscription = pool.sub({cb: ..., filter: ....}) const mySubscription = pool.sub({filter: ...., skipVerification: false, beforeSend: ....})
mySubscription.sub({filter: ....}) mySubscription.sub({filter: ....})
mySubscription.sub({cb: ...}) mySubscription.sub({skipVerification: true})
// listen for messages for subscription
mySubscription.on('event', onEvent)
mySubscription.on('eose', onEose)
// close subscription
mySubscription.unsub() mySubscription.unsub()
// get specific event // get specific event
const specificChannel = pool.sub({ const specificChannel = pool.sub({ filter: {id: '<hex>'}})
cb: (event, relay) => { .on('event', ({event, relay}) => {
console.log('got specific event from relay', event, relay) console.log('got specific event from relay', event, relay)
specificChannel.unsub() specificChannel.unsub()
}, })
filter: {id: '<hex>'}
})
// or get a specific event plus all the events that reference it in the 'e' tag // or get a specific event plus all the events that reference it in the 'e' tag
pool.sub({ cb: (event, relay) => { ... }, filter: [{id: '<hex>'}, {'#e': '<hex>'}] }) pool.sub({ filter: [{id: '<hex>'}, {'#e': '<hex>'}] })
// get all events // get all events
pool.sub({cb: (event, relay) => {...}, filter: {}}) pool.sub({ filter: {} })
// get recent events // get recent events
pool.sub({cb: (event, relay) => {...}, filter: {since: timestamp}}) pool.sub({ filter: {since: timestamp} })
// publishing events(inside an async function): // publishing events(inside an async function):
const ev = await pool.publish(eventObject, (status, url) => { const ev = await pool.publish(eventObject, (status, url) => {

11
index.js
View File

@@ -1,6 +1,6 @@
import {generatePrivateKey, getPublicKey} from './keys.js' import { generatePrivateKey, getPublicKey } from './keys.js'
import {relayConnect} from './relay.js' import { relayInit } from './relay.js'
import {relayPool} from './pool.js' import { relayPool } from './pool.js'
import { import {
getBlankEvent, getBlankEvent,
signEvent, signEvent,
@@ -9,11 +9,11 @@ import {
serializeEvent, serializeEvent,
getEventHash getEventHash
} from './event.js' } from './event.js'
import {matchFilter, matchFilters} from './filter.js' import { matchFilter, matchFilters } from './filter.js'
export { export {
generatePrivateKey, generatePrivateKey,
relayConnect, relayInit,
relayPool, relayPool,
signEvent, signEvent,
validateEvent, validateEvent,
@@ -25,3 +25,4 @@ export {
matchFilter, matchFilter,
matchFilters matchFilters
} }

197
pool.js
View File

@@ -1,5 +1,5 @@
import {getEventHash, verifySignature, signEvent} from './event.js' import { getEventHash, verifySignature, signEvent } from './event.js'
import {relayConnect, normalizeRelayURL} from './relay.js' import { relayInit, normalizeRelayURL } from './relay.js'
export function relayPool() { export function relayPool() {
var globalPrivateKey var globalPrivateKey
@@ -14,88 +14,107 @@ export function relayPool() {
// been published -- or at least attempted to be published -- to all relays // been published -- or at least attempted to be published -- to all relays
wait: false wait: false
} }
//map with all the relays where the url is the id // map with all the relays where the url is the id
//Map<string,{relay:Relay,policy:RelayPolicy> // Map<string,{relay:Relay,policy:RelayPolicy>
const relays = {} const relays = {}
const noticeCallbacks = [] const openSubs = {}
function propagateNotice(notice, relayURL) {
for (let i = 0; i < noticeCallbacks.length; i++) {
let {relay} = relays[relayURL]
noticeCallbacks[i](notice, relay)
}
}
const activeSubscriptions = {} const activeSubscriptions = {}
const poolListeners = { notice: [], connection: [], disconnection: [], error: [] }
//sub creates a Subscription object {sub:Function, unsub:Function, addRelay:Function,removeRelay :Function } // sub creates a Subscription object {sub:Function, unsub:Function, addRelay:Function,removeRelay :Function }
const sub = ({cb, filter, beforeSend}, id, cbEose) => { const sub = ({ filter, beforeSend, skipVerification }, id) => {
//check if it has an id, if not assign one // check if it has an id, if not assign one
if (!id) id = Math.random().toString().slice(2) if (!id) id = Math.random().toString().slice(2)
// save sub settings
openSubs[id] = {
filter,
beforeSend,
skipVerification,
}
const subListeners = { event: [], eose: [] }
const subControllers = Object.fromEntries( const subControllers = Object.fromEntries(
//Convert the map<string,Relay> to a Relay[] // Convert the map<string,Relay> to a Relay[]
Object.values(relays) Object.values(relays)
//takes only relays that can be read // takes only relays that can be read
.filter(({policy}) => policy.read) .filter(({ policy }) => policy.read)
//iterate all the rellies and create the array [url:string,sub:SubscriptionCallback] // iterate all the relays and create the array [url:string,sub:SubscriptionCallback, listeners]
.map(({relay}) => [ .map(({ relay }) => [
relay.url, relay.url,
relay.sub({cb: event => cb(event, relay.url), filter, beforeSend}, id, relay.sub(openSubs[id], id),
() => cbEose(relay.url))
]) ])
) )
const activeCallback = cb //assign callback for the suscription // Unsub deletes itself
const activeFilters = filter //assigng filter for the suscrition
const activeBeforeSend = beforeSend //assign before send fucntion
//Unsub deletes itself
const unsub = () => { const unsub = () => {
//iterate the map of subControllers and call the unsub function of it relays // iterate the map of subControllers and call the unsub function of it relays
Object.values(subControllers).forEach(sub => sub.unsub()) Object.values(subControllers).forEach(sub => sub.unsub())
delete activeSubscriptions[id] delete openSubs[id]
delete activeSubscriptions[id]
} }
const sub = ({ const sub = ({
cb = activeCallback, filter = openSubs[id].filter,
filter = activeFilters, beforeSend = openSubs[id].beforeSend,
beforeSend = activeBeforeSend skipVerification = openSubs[id].skipVerification }
}) => { ) => {
//Iterates the subControllers and add them the atributes of our Subscription (callback, filter,etc.) // update sub settings
Object.entries(subControllers).map(([relayURL, sub]) => [ openSubs[id] = {
relayURL, filter,
sub.sub({cb: event => cb(event, relayURL), filter, beforeSend}, id, beforeSend,
() => cbEose(relayURL)) skipVerification,
]) }
//returns the current suscripcion // update relay subs
Object.entries(subControllers).forEach(([relayURL, sub]) => {
sub.sub(openSubs[id], id)
})
// returns the current suscripcion
return activeSubscriptions[id] return activeSubscriptions[id]
} }
//addRelay adds a relay to the subControllers map so the current subscription can use it // addRelay adds a relay to the subControllers map so the current subscription can use it
const addRelay = relay => { const addRelay = relay => {
subControllers[relay.url] = relay.sub( for (let type of Object.keys(subListeners)) {
{cb: event => cb(event, relay.url), filter, beforeSend}, if (subListeners[type].length) subListeners[type].forEach(cb => relay.on(type, cb, id))
id, () => cbEose(relay.url) }
) subControllers[relay.url] = relay.sub(openSubs[id], id)
return activeSubscriptions[id] return activeSubscriptions[id]
} }
//removeRelay deletes a relay from the subControllers map, it also handles the unsubscription from the relay // removeRelay deletes a relay from the subControllers map, it also handles the unsubscription from the relay
const removeRelay = relayURL => { const removeRelay = relayURL => {
if (relayURL in subControllers) { if (relayURL in subControllers) {
subControllers[relayURL].unsub() subControllers[relayURL].unsub()
delete subControllers[relayURL]
if (Object.keys(subControllers).length === 0) unsub() if (Object.keys(subControllers).length === 0) unsub()
} }
return activeSubscriptions[id] return activeSubscriptions[id]
} }
// on creates listener for sub ('EVENT', 'EOSE', etc)
const on = (type, cb) => {
subListeners[type].push(cb)
Object.values(relays).filter(({ policy }) => policy.read).forEach(({ relay }) => relay.on(type, cb, id))
return activeSubscriptions[id]
}
// off destroys listener for sub ('EVENT', 'EOSE', etc)
const off = (type, cb) => {
if (!subListeners[type].length) return
let index = subListeners[type].indexOf(cb)
if (index !== -1) subListeners[type].splice(index, 1)
Object.values(relays).forEach(({ relay }) => relay.off(type, cb, id))
return activeSubscriptions[id]
}
//add the object created to activeSubscriptions map // add the object created to activeSubscriptions map
activeSubscriptions[id] = { activeSubscriptions[id] = {
sub, sub,
unsub, unsub,
addRelay, addRelay,
removeRelay removeRelay,
on,
off
} }
return activeSubscriptions[id] return activeSubscriptions[id]
@@ -113,64 +132,72 @@ export function relayPool() {
setPolicy(key, value) { setPolicy(key, value) {
poolPolicy[key] = value poolPolicy[key] = value
}, },
//addRelay adds a relay to the pool and to all its subscriptions // addRelay adds a relay to the pool and to all its subscriptions
addRelay(url, policy = {read: true, write: true}) { addRelay(url, policy = { read: true, write: true }) {
let relayURL = normalizeRelayURL(url) let relayURL = normalizeRelayURL(url)
if (relayURL in relays) return if (relayURL in relays) return
let relay = relayConnect(url, notice => { let relay = relayInit(url)
propagateNotice(notice, relayURL)
}) for (let type of Object.keys(poolListeners)) {
relays[relayURL] = {relay, policy} let cbs = poolListeners[type] || []
if (cbs.length) poolListeners[type].forEach(cb => relay.on(type, cb))
}
if (policy.read) { if (policy.read) {
Object.values(activeSubscriptions).forEach(subscription => Object.values(activeSubscriptions).forEach(sub => sub.addRelay(relay))
subscription.addRelay(relay)
)
} }
relay.connect()
relays[relayURL] = { relay, policy }
return relay return relay
}, },
//remove relay deletes the relay from the pool and from all its subscriptions // remove relay deletes the relay from the pool and from all its subscriptions
removeRelay(url) { removeRelay(url) {
let relayURL = normalizeRelayURL(url) let relayURL = normalizeRelayURL(url)
let data = relays[relayURL] let data = relays[relayURL]
if (!data) return if (!data) return
let {relay} = data let { relay } = data
Object.values(activeSubscriptions).forEach(subscription => Object.values(activeSubscriptions).forEach(sub => sub.removeRelay(relayURL))
subscription.removeRelay(relay)
)
relay.close() relay.close()
delete relays[relayURL] delete relays[relayURL]
}, },
//getRelayList return an array with all the relays stored // getRelayList return an array with all the relays stored
getRelayList() { getRelayList() {
return Object.values(relays) return Object.values(relays)
}, },
relayChangePolicy(url, policy = {read: true, write: true}) { relayChangePolicy(url, policy = { read: true, write: true }) {
let relayURL = normalizeRelayURL(url) let relayURL = normalizeRelayURL(url)
let data = relays[relayURL] let data = relays[relayURL]
if (!data) return if (!data) return
relays[relayURL].policy = policy let { relay } = data
if (relays[relayURL].policy.read === true && policy.read === false)
Object.values(activeSubscriptions).forEach(sub => sub.removeRelay(relayURL))
else if (relays[relayURL].policy.read === false && policy.read === true)
Object.values(activeSubscriptions).forEach(sub => sub.addRelay(relay));
relays[relayURL].policy = policy
return relays[relayURL] return relays[relayURL]
}, },
onNotice(cb) { on(type, cb) {
noticeCallbacks.push(cb) poolListeners[type] = poolListeners[type] || []
poolListeners[type].push(cb)
Object.values(relays).forEach(({ relay }) => relay.on(type, cb))
}, },
offNotice(cb) { off(type, cb) {
let index = noticeCallbacks.indexOf(cb) let index = poolListeners[type].indexOf(cb)
if (index !== -1) noticeCallbacks.splice(index, 1) if (index !== -1) poolListeners[type].splice(index, 1)
Object.values(relays).forEach(({ relay }) => relay.off(type, cb))
}, },
//publish send a event to the relays // publish send a event to the relays
async publish(event, statusCallback) { async publish(event, statusCallback) {
event.id = getEventHash(event) event.id = getEventHash(event)
//if the event is not signed then sign it // if the event is not signed then sign it
if (!event.sig) { if (!event.sig) {
event.tags = event.tags || [] event.tags = event.tags || []
@@ -195,9 +222,9 @@ export function relayPool() {
} }
} }
//get the writable relays // get the writable relays
let writeable = Object.values(relays) let writeable = Object.values(relays)
.filter(({policy}) => policy.write) .filter(({ policy }) => policy.write)
.sort(() => Math.random() - 0.5) // random .sort(() => Math.random() - 0.5) // random
let maxTargets = poolPolicy.randomChoice let maxTargets = poolPolicy.randomChoice
@@ -206,10 +233,10 @@ export function relayPool() {
let successes = 0 let successes = 0
//if the pool policy set to want until event send // if the pool policy set to wait until event send
if (poolPolicy.wait) { if (poolPolicy.wait) {
for (let i = 0; i < writeable.length; i++) { for (let i = 0; i < writeable.length; i++) {
let {relay} = writeable[i] let { relay } = writeable[i]
try { try {
await new Promise(async (resolve, reject) => { await new Promise(async (resolve, reject) => {
@@ -231,9 +258,9 @@ export function relayPool() {
/***/ /***/
} }
} }
//if the pool policy dont want to wait until event send // if the pool policy dont want to wait until event send
} else { } else {
writeable.forEach(async ({relay}) => { writeable.forEach(async ({ relay }) => {
let callback = statusCallback let callback = statusCallback
? status => statusCallback(status, relay.url) ? status => statusCallback(status, relay.url)
: null : null

186
relay.js
View File

@@ -2,8 +2,8 @@
import 'websocket-polyfill' import 'websocket-polyfill'
import {verifySignature, validateEvent} from './event.js' import { verifySignature, validateEvent } from './event.js'
import {matchFilters} from './filter.js' import { matchFilters } from './filter.js'
export function normalizeRelayURL(url) { export function normalizeRelayURL(url) {
let [host, ...qs] = url.trim().split('?') let [host, ...qs] = url.trim().split('?')
@@ -13,12 +13,19 @@ export function normalizeRelayURL(url) {
return [host, ...qs].join('?') return [host, ...qs].join('?')
} }
export function relayConnect(url, onNotice = () => {}, onError = () => {}) { export function relayInit(url) {
url = normalizeRelayURL(url) let relay = normalizeRelayURL(url) // set relay url
var ws, resolveOpen, untilOpen, wasClosed var ws, resolveOpen, untilOpen, wasClosed, closed
var openSubs = {} var openSubs = {}
var isSetToSkipVerification = {} var listeners = {
event: { '_': [] },
eose: { '_': [] },
connection: { '_': [] },
disconnection: { '_': [] },
error: { '_': [] },
notice: { '_': [] },
}
let attemptNumber = 1 let attemptNumber = 1
let nextAttemptSeconds = 1 let nextAttemptSeconds = 1
@@ -28,32 +35,27 @@ export function relayConnect(url, onNotice = () => {}, onError = () => {}) {
}) })
} }
var eventListeners = {} function connectRelay() {
var eoseListeners = {} ws = new WebSocket(relay)
function connect() {
ws = new WebSocket(url)
ws.onopen = () => { ws.onopen = () => {
console.log('connected to', url) listeners.connection._.forEach(cb => cb({ type: 'connection', relay }))
resolveOpen() resolveOpen()
// restablish old subscriptions // restablish old subscriptions
if (wasClosed) { if (wasClosed) {
wasClosed = false wasClosed = false
for (let channel in openSubs) { for (let id in openSubs) {
let filters = openSubs[channel] sub(openSubs[id], id)
let eventCb = eventListeners[channel]
let eoseCb = eoseListeners[channel]
sub({eventCb, filter: filters}, channel, eoseCb)
} }
} }
} }
ws.onerror = err => { ws.onerror = error => {
console.log('error connecting to relay', url) listeners.error._.forEach(cb => cb({ type: 'error', relay, error }))
onError(err)
} }
ws.onclose = () => { ws.onclose = async () => {
listeners.disconnection._.forEach(cb => cb({ type: 'disconnection', relay }))
if (closed) return
resetOpenState() resetOpenState()
attemptNumber++ attemptNumber++
nextAttemptSeconds += attemptNumber ** 3 nextAttemptSeconds += attemptNumber ** 3
@@ -61,12 +63,12 @@ export function relayConnect(url, onNotice = () => {}, onError = () => {}) {
nextAttemptSeconds = 14400 // 4 hours nextAttemptSeconds = 14400 // 4 hours
} }
console.log( console.log(
`relay ${url} connection closed. reconnecting in ${nextAttemptSeconds} seconds.` `relay ${relay} connection closed. reconnecting in ${nextAttemptSeconds} seconds.`
) )
setTimeout(async () => { setTimeout(async () => {
try { try {
connect() connectRelay()
} catch (err) {} } catch (err) { }
}, nextAttemptSeconds * 1000) }, nextAttemptSeconds * 1000)
wasClosed = true wasClosed = true
@@ -82,38 +84,33 @@ export function relayConnect(url, onNotice = () => {}, onError = () => {}) {
if (data.length >= 1) { if (data.length >= 1) {
switch (data[0]) { switch (data[0]) {
case 'NOTICE':
if (data.length !== 2) {
// ignore empty or malformed notice
return
}
console.log(`message from relay ${url}: ${data[1]}`)
onNotice(data[1])
return
case 'EOSE':
if (data.length !== 2) {
// ignore malformed EOSE
return
}
console.log(`Channel ${data[1]}: End-of-stored-events`)
if (eoseListeners[data[1]]) {
eoseListeners[data[1]]()
}
return
case 'EVENT': case 'EVENT':
if (data.length !== 3) { if (data.length !== 3) return // ignore empty or malformed EVENT
// ignore malformed EVENT
return let id = data[1]
}
let channel = data[1]
let event = data[2] let event = data[2]
if (validateEvent(event) && if (validateEvent(event) && openSubs[id] &&
(isSetToSkipVerification[channel] || verifySignature(event)) && (openSubs[id].skipVerification || verifySignature(event)) &&
eventListeners[channel] && matchFilters(openSubs[id].filter, event)
matchFilters(openSubs[channel], event)) { ) {
eventListeners[channel](event) if (listeners.event[id]?.length) listeners.event[id].forEach(cb => cb({ type: 'event', relay, id, event }))
if (listeners.event._.length) listeners.event._.forEach(cb => cb({ type: 'event', relay, id, event }))
} }
return return
case 'EOSE': {
if (data.length !== 2) return // ignore empty or malformed EOSE
let id = data[1]
if (listeners.eose[id]?.length) listeners.eose[data[1]].forEach(cb => cb({ type: 'eose', relay, id }))
if (listeners.eose._.length) listeners.eose._.forEach(cb => cb({ type: 'eose', relay, id }))
return
}
case 'NOTICE':
if (data.length !== 2) return // ignore empty or malformed NOTICE
let notice = data[1]
if (listeners.notice._.length) listeners.notice._.forEach(cb => cb({ type: 'notice', relay, notice }))
return
} }
} }
} }
@@ -121,9 +118,12 @@ export function relayConnect(url, onNotice = () => {}, onError = () => {}) {
resetOpenState() resetOpenState()
try { async function connect() {
connect() if (ws?.readyState && ws.readyState === 1) return // ws already open
} catch (err) {} try {
connectRelay()
} catch (err) { }
}
async function trySend(params) { async function trySend(params) {
let msg = JSON.stringify(params) let msg = JSON.stringify(params)
@@ -132,75 +132,79 @@ export function relayConnect(url, onNotice = () => {}, onError = () => {}) {
ws.send(msg) ws.send(msg)
} }
const sub = ( const sub = ({ filter, beforeSend, skipVerification }, id = Math.random().toString().slice(2)) => {
{cb, filter, beforeSend, skipVerification},
channel = Math.random().toString().slice(2),
eoseCb
) => {
var filters = [] var filters = []
if (Array.isArray(filter)) { if (Array.isArray(filter)) {
filters = filter filters = filter
} else { } else {
filters.push(filter) filters.push(filter)
} }
filter = filters
if (beforeSend) { if (beforeSend) {
const beforeSendResult = beforeSend({filter, relay: url, channel}) const beforeSendResult = beforeSend({ filter, relay, id })
filters = beforeSendResult.filter filter = beforeSendResult.filter
} }
trySend(['REQ', channel, ...filters]) openSubs[id] = {
eventListeners[channel] = cb filter,
eoseListeners[channel] = eoseCb beforeSend,
openSubs[channel] = filters skipVerification,
isSetToSkipVerification[channel] = skipVerification }
trySend(['REQ', id, ...filter])
const activeCallback = cb
const activeFilters = filters
const activeBeforeSend = beforeSend
return { return {
sub: ({ sub: ({
cb = activeCallback, filter = openSubs[id].filter,
filter = activeFilters, beforeSend = openSubs[id].beforeSend,
beforeSend = activeBeforeSend skipVerification = openSubs[id].skipVerification }
}) => sub({cb, filter, beforeSend, skipVerification}, channel, eoseCb), ) => sub({ filter, beforeSend, skipVerification }, id),
unsub: () => { unsub: () => {
delete openSubs[channel] delete openSubs[id]
delete eventListeners[channel] delete listeners.event[id]
delete eoseListeners[channel] delete listeners.eose[id]
delete isSetToSkipVerification[channel] trySend(['CLOSE', id])
trySend(['CLOSE', channel])
} }
} }
} }
function on(type, cb, id = '_') {
listeners[type][id] = listeners[type][id] || []
listeners[type][id].push(cb)
}
function off(type, cb, id = '_') {
if (!listeners[type][id].length) return
let index = listeners[type][id].indexOf(cb)
if (index !== -1) listeners[type][id].splice(index, 1)
}
return { return {
url, url,
sub, sub,
on,
off,
async publish(event, statusCallback) { async publish(event, statusCallback) {
try { try {
await trySend(['EVENT', event]) await trySend(['EVENT', event])
if (statusCallback) { if (statusCallback) {
let id = `monitor-${event.id.slice(0, 5)}`
statusCallback(0) statusCallback(0)
let {unsub} = sub( let { unsub } = sub({ filter: { ids: [event.id] } }, id)
{ on('event', () => {
cb: () => { statusCallback(1)
statusCallback(1) unsub()
unsub() clearTimeout(willUnsub)
clearTimeout(willUnsub) }, id)
},
filter: {ids: [event.id]}
},
`monitor-${event.id.slice(0, 5)}`
)
let willUnsub = setTimeout(unsub, 5000) let willUnsub = setTimeout(unsub, 5000)
} }
} catch (err) { } catch (err) {
if (statusCallback) statusCallback(-1) if (statusCallback) statusCallback(-1)
} }
}, },
connect,
close() { close() {
closed = true // prevent ws from trying to reconnect
ws.close() ws.close()
}, },
get status() { get status() {