mirror of
https://github.com/nbd-wtf/nostr-tools.git
synced 2025-12-09 00:28:51 +00:00
rewrite pool.ts to be much simpler.
This commit is contained in:
fiatjaf
314
pool.ts
314
pool.ts
@@ -1,237 +1,167 @@
|
|||||||
import { eventsGenerator, relayInit, type Relay, type Sub, type SubscriptionOptions } from './relay.ts'
|
import { relayConnect, type Relay, SubscriptionParams, Subscription } from './relay.ts'
|
||||||
import { normalizeURL } from './utils.ts'
|
import { normalizeURL } from './utils.ts'
|
||||||
|
|
||||||
import type { Event } from './event.ts'
|
import type { Event } from './event.ts'
|
||||||
import { matchFilters, mergeFilters, type Filter } from './filter.ts'
|
import { type Filter } from './filter.ts'
|
||||||
|
|
||||||
type BatchedRequest = {
|
export type SubscribeManyParams = Omit<SubscriptionParams, 'onclose'> & {
|
||||||
filters: Filter[]
|
eoseSubTimeout: number
|
||||||
relays: string[]
|
onclose?: (reasons: string[]) => void
|
||||||
resolve: (events: Event[]) => void
|
|
||||||
events: Event[]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export class SimplePool {
|
export class SimplePool {
|
||||||
private _conn: { [url: string]: Relay }
|
private relays = new Map<string, Relay>()
|
||||||
private _seenOn: { [id: string]: Set<string> } = {} // a map of all events we've seen in each relay
|
|
||||||
private batchedByKey: { [batchKey: string]: BatchedRequest[] } = {}
|
|
||||||
|
|
||||||
private eoseSubTimeout: number
|
|
||||||
private getTimeout: number
|
|
||||||
private seenOnEnabled: boolean = true
|
|
||||||
private batchInterval: number = 100
|
|
||||||
|
|
||||||
constructor(
|
|
||||||
options: {
|
|
||||||
eoseSubTimeout?: number
|
|
||||||
getTimeout?: number
|
|
||||||
seenOnEnabled?: boolean
|
|
||||||
batchInterval?: number
|
|
||||||
} = {},
|
|
||||||
) {
|
|
||||||
this._conn = {}
|
|
||||||
this.eoseSubTimeout = options.eoseSubTimeout || 3400
|
|
||||||
this.getTimeout = options.getTimeout || 3400
|
|
||||||
this.seenOnEnabled = options.seenOnEnabled !== false
|
|
||||||
this.batchInterval = options.batchInterval || 100
|
|
||||||
}
|
|
||||||
|
|
||||||
close(relays: string[]): void {
|
|
||||||
relays.forEach(url => {
|
|
||||||
let relay = this._conn[normalizeURL(url)]
|
|
||||||
if (relay) relay.close()
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
async ensureRelay(url: string): Promise<Relay> {
|
async ensureRelay(url: string): Promise<Relay> {
|
||||||
const nm = normalizeURL(url)
|
url = normalizeURL(url)
|
||||||
|
|
||||||
if (!this._conn[nm]) {
|
let relay = this.relays.get(url)
|
||||||
this._conn[nm] = relayInit(nm, {
|
if (!relay) {
|
||||||
getTimeout: this.getTimeout * 0.9,
|
relay = relayConnect(url)
|
||||||
listTimeout: this.getTimeout * 0.9,
|
this.relays.set(url, relay)
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const relay = this._conn[nm]
|
|
||||||
await relay.connect()
|
|
||||||
return relay
|
return relay
|
||||||
}
|
}
|
||||||
|
|
||||||
sub(relays: string[], filters: Filter[], opts?: SubscriptionOptions): Sub {
|
async subscribeMany(
|
||||||
let _knownIds: Set<string> = new Set()
|
relays: string[],
|
||||||
let modifiedOpts = { ...(opts || {}) }
|
filters: Filter[],
|
||||||
modifiedOpts.alreadyHaveEvent = (id, url) => {
|
params: SubscribeManyParams,
|
||||||
if (opts?.alreadyHaveEvent?.(id, url)) {
|
): Promise<{ close: () => void }> {
|
||||||
|
const _knownIds = new Set<string>()
|
||||||
|
params.alreadyHaveEvent = (id: string) => {
|
||||||
|
if (params.alreadyHaveEvent?.(id)) {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
if (this.seenOnEnabled) {
|
const have = _knownIds.has(id)
|
||||||
let set = this._seenOn[id] || new Set()
|
_knownIds.add(id)
|
||||||
set.add(url)
|
return have
|
||||||
this._seenOn[id] = set
|
|
||||||
}
|
|
||||||
return _knownIds.has(id)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
let subs: Sub[] = []
|
const subs: Subscription[] = []
|
||||||
let eventListeners: Set<any> = new Set()
|
|
||||||
let eoseListeners: Set<() => void> = new Set()
|
// batch all EOSEs into a single
|
||||||
let eosesMissing = relays.length
|
let eosesMissing = relays.length
|
||||||
|
let handleEose = () => {
|
||||||
|
eosesMissing--
|
||||||
|
if (eosesMissing === 0) {
|
||||||
|
clearTimeout(eoseTimeout)
|
||||||
|
params.oneose?.()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const eoseTimeout = setTimeout(() => {
|
||||||
|
handleEose = () => {}
|
||||||
|
params.oneose?.()
|
||||||
|
}, params.eoseSubTimeout || 5400)
|
||||||
|
|
||||||
let eoseSent = false
|
// batch all closes into a single
|
||||||
let eoseTimeout = setTimeout(
|
const closesReceived: string[] = []
|
||||||
() => {
|
const handleClose = (i: number, reason: string) => {
|
||||||
eoseSent = true
|
handleEose()
|
||||||
for (let cb of eoseListeners.values()) cb()
|
closesReceived[i] = reason
|
||||||
},
|
if (closesReceived.length === relays.length) {
|
||||||
opts?.eoseSubTimeout || this.eoseSubTimeout,
|
params.onclose?.(closesReceived)
|
||||||
)
|
}
|
||||||
|
}
|
||||||
|
|
||||||
relays
|
// open a subscription in all given relays
|
||||||
.filter((r, i, a) => a.indexOf(r) === i)
|
await Promise.all(
|
||||||
.forEach(async relay => {
|
relays.map(normalizeURL).map(async (url, i) => {
|
||||||
let r
|
if (relays.indexOf(url) !== i) {
|
||||||
|
// duplicate
|
||||||
|
handleClose(i, 'duplicate')
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
let relay: Relay
|
||||||
try {
|
try {
|
||||||
r = await this.ensureRelay(relay)
|
relay = await this.ensureRelay(url)
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
handleEose()
|
handleEose()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if (!r) return
|
|
||||||
let s = r.sub(filters, modifiedOpts)
|
|
||||||
s.on('event', event => {
|
|
||||||
_knownIds.add(event.id as string)
|
|
||||||
for (let cb of eventListeners.values()) cb(event)
|
|
||||||
})
|
|
||||||
s.on('eose', () => {
|
|
||||||
if (eoseSent) return
|
|
||||||
handleEose()
|
|
||||||
})
|
|
||||||
subs.push(s)
|
|
||||||
|
|
||||||
function handleEose() {
|
let subscription = await relay.subscribe(filters, {
|
||||||
eosesMissing--
|
...params,
|
||||||
if (eosesMissing === 0) {
|
oneose: handleEose,
|
||||||
clearTimeout(eoseTimeout)
|
onclose: reason => handleClose(i, reason),
|
||||||
for (let cb of eoseListeners.values()) cb()
|
})
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
let greaterSub: Sub = {
|
subs.push(subscription)
|
||||||
sub(filters, opts) {
|
}),
|
||||||
subs.forEach(sub => sub.sub(filters, opts))
|
)
|
||||||
return greaterSub as any
|
|
||||||
},
|
return {
|
||||||
unsub() {
|
close() {
|
||||||
subs.forEach(sub => sub.unsub())
|
subs.forEach(sub => {
|
||||||
},
|
sub.close()
|
||||||
on(type, cb) {
|
})
|
||||||
if (type === 'event') {
|
|
||||||
eventListeners.add(cb)
|
|
||||||
} else if (type === 'eose') {
|
|
||||||
eoseListeners.add(cb as () => void | Promise<void>)
|
|
||||||
}
|
|
||||||
},
|
|
||||||
off(type, cb) {
|
|
||||||
if (type === 'event') {
|
|
||||||
eventListeners.delete(cb)
|
|
||||||
} else if (type === 'eose') eoseListeners.delete(cb as () => void | Promise<void>)
|
|
||||||
},
|
|
||||||
get events() {
|
|
||||||
return eventsGenerator(greaterSub)
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
return greaterSub
|
|
||||||
}
|
}
|
||||||
|
|
||||||
get(relays: string[], filter: Filter, opts?: SubscriptionOptions): Promise<Event | null> {
|
async subscribeManyEose(
|
||||||
return new Promise(resolve => {
|
relays: string[],
|
||||||
let sub = this.sub(relays, [filter], opts)
|
filters: Filter[],
|
||||||
let timeout = setTimeout(() => {
|
params: Pick<SubscribeManyParams, 'id' | 'onevent' | 'onclose' | 'eoseSubTimeout'>,
|
||||||
sub.unsub()
|
): Promise<{ close: () => void }> {
|
||||||
resolve(null)
|
const sub = await this.subscribeMany(relays, filters, {
|
||||||
}, this.getTimeout)
|
...params,
|
||||||
sub.on('event', event => {
|
oneose() {
|
||||||
resolve(event)
|
sub.close()
|
||||||
clearTimeout(timeout)
|
},
|
||||||
sub.unsub()
|
})
|
||||||
|
return sub
|
||||||
|
}
|
||||||
|
|
||||||
|
get(
|
||||||
|
relays: string[],
|
||||||
|
filter: Filter,
|
||||||
|
params: Pick<SubscribeManyParams, 'id' | 'eoseSubTimeout'>,
|
||||||
|
): Promise<Event | null> {
|
||||||
|
return new Promise(async (resolve, reject) => {
|
||||||
|
const sub = await this.subscribeManyEose(relays, [filter], {
|
||||||
|
...params,
|
||||||
|
onevent(event: Event) {
|
||||||
|
resolve(event)
|
||||||
|
sub.close()
|
||||||
|
},
|
||||||
|
onclose(reasons: string[]) {
|
||||||
|
const err = new Error('subscriptions closed')
|
||||||
|
err.cause = reasons
|
||||||
|
reject(err)
|
||||||
|
},
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
list(relays: string[], filters: Filter[], opts?: SubscriptionOptions): Promise<Event[]> {
|
publish(relays: string[], event: Event): Promise<string>[] {
|
||||||
return new Promise(resolve => {
|
return relays.map(normalizeURL).map(async (url, i) => {
|
||||||
let events: Event[] = []
|
if (relays.indexOf(url) !== i) {
|
||||||
let sub = this.sub(relays, filters, opts)
|
// duplicate
|
||||||
|
return Promise.reject('duplicate')
|
||||||
sub.on('event', event => {
|
|
||||||
events.push(event)
|
|
||||||
})
|
|
||||||
|
|
||||||
// we can rely on an eose being emitted here because pool.sub() will fake one
|
|
||||||
sub.on('eose', () => {
|
|
||||||
sub.unsub()
|
|
||||||
resolve(events)
|
|
||||||
})
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
batchedList(batchKey: string, relays: string[], filters: Filter[]): Promise<Event[]> {
|
|
||||||
return new Promise(resolve => {
|
|
||||||
if (!this.batchedByKey[batchKey]) {
|
|
||||||
this.batchedByKey[batchKey] = [
|
|
||||||
{
|
|
||||||
filters,
|
|
||||||
relays,
|
|
||||||
resolve,
|
|
||||||
events: [],
|
|
||||||
},
|
|
||||||
]
|
|
||||||
|
|
||||||
setTimeout(() => {
|
|
||||||
Object.keys(this.batchedByKey).forEach(async batchKey => {
|
|
||||||
const batchedRequests = this.batchedByKey[batchKey]
|
|
||||||
|
|
||||||
const filters = [] as Filter[]
|
|
||||||
const relays = [] as string[]
|
|
||||||
batchedRequests.forEach(br => {
|
|
||||||
filters.push(...br.filters)
|
|
||||||
relays.push(...br.relays)
|
|
||||||
})
|
|
||||||
|
|
||||||
const sub = this.sub(relays, [mergeFilters(...filters)])
|
|
||||||
sub.on('event', event => {
|
|
||||||
batchedRequests.forEach(br => matchFilters(br.filters, event) && br.events.push(event))
|
|
||||||
})
|
|
||||||
sub.on('eose', () => {
|
|
||||||
sub.unsub()
|
|
||||||
batchedRequests.forEach(br => br.resolve(br.events))
|
|
||||||
})
|
|
||||||
|
|
||||||
delete this.batchedByKey[batchKey]
|
|
||||||
})
|
|
||||||
}, this.batchInterval)
|
|
||||||
} else {
|
|
||||||
this.batchedByKey[batchKey].push({
|
|
||||||
filters,
|
|
||||||
relays,
|
|
||||||
resolve,
|
|
||||||
events: [],
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
publish(relays: string[], event: Event): Promise<void>[] {
|
let r = await this.ensureRelay(url)
|
||||||
return relays.map(async relay => {
|
|
||||||
let r = await this.ensureRelay(relay)
|
|
||||||
return r.publish(event)
|
return r.publish(event)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
seenOn(id: string): string[] {
|
export class RelayTrackingPool extends SimplePool {
|
||||||
return Array.from(this._seenOn[id]?.values?.() || [])
|
public seenOn = new Map<string, Set<Relay>>()
|
||||||
|
|
||||||
|
subscribeMany(relays: string[], filters: Filter[], params: SubscribeManyParams): Promise<{ close: () => void }> {
|
||||||
|
params.receivedEvent = (relay: Relay, id: string) => {
|
||||||
|
let set = this.seenOn.get(id)
|
||||||
|
if (!set) {
|
||||||
|
set = new Set()
|
||||||
|
this.seenOn.set(id, set)
|
||||||
|
}
|
||||||
|
set.add(relay)
|
||||||
|
}
|
||||||
|
|
||||||
|
return super.subscribeMany(relays, filters, params)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
13
relay.ts
13
relay.ts
@@ -128,8 +128,13 @@ export class Relay {
|
|||||||
// we do this before parsing the JSON to not have to do that for duplicate events
|
// we do this before parsing the JSON to not have to do that for duplicate events
|
||||||
// since JSON parsing is slow
|
// since JSON parsing is slow
|
||||||
const id = getHex64(json, 'id')
|
const id = getHex64(json, 'id')
|
||||||
so.receivedEvent?.(id) // this is so the client knows this relay had this event
|
const alreadyHave = so.alreadyHaveEvent?.(id)
|
||||||
if (so.alreadyHaveEvent?.(id)) {
|
|
||||||
|
// notify any interested client that the relay has this event
|
||||||
|
// (do this after alreadyHaveEvent() because the client may rely on this to answer that)
|
||||||
|
so.receivedEvent?.(this, id)
|
||||||
|
|
||||||
|
if (alreadyHave) {
|
||||||
// if we had already seen this event we can just stop here
|
// if we had already seen this event we can just stop here
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -263,7 +268,7 @@ export class Subscription {
|
|||||||
public eosed: boolean = false
|
public eosed: boolean = false
|
||||||
|
|
||||||
public alreadyHaveEvent: ((id: string) => boolean) | undefined
|
public alreadyHaveEvent: ((id: string) => boolean) | undefined
|
||||||
public receivedEvent: ((id: string) => boolean) | undefined
|
public receivedEvent: ((relay: Relay, id: string) => void) | undefined
|
||||||
public readonly filters: Filter[]
|
public readonly filters: Filter[]
|
||||||
|
|
||||||
public onevent: (evt: Event) => void
|
public onevent: (evt: Event) => void
|
||||||
@@ -298,7 +303,7 @@ export type SubscriptionParams = {
|
|||||||
oneose?: () => void
|
oneose?: () => void
|
||||||
onclose?: (reason: string) => void
|
onclose?: (reason: string) => void
|
||||||
alreadyHaveEvent?: (id: string) => boolean
|
alreadyHaveEvent?: (id: string) => boolean
|
||||||
receivedEvent?: (id: string) => boolean
|
receivedEvent?: (relay: Relay, id: string) => void
|
||||||
}
|
}
|
||||||
|
|
||||||
export type CountResolver = {
|
export type CountResolver = {
|
||||||
|
|||||||
Reference in New Issue
Block a user