mirror of
https://github.com/nbd-wtf/nostr-tools.git
synced 2025-12-09 08:38:50 +00:00
add batchedList method to SimplePool
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "nostr-tools",
|
"name": "nostr-tools",
|
||||||
"version": "1.14.1",
|
"version": "1.14.2",
|
||||||
"description": "Tools for making a Nostr client.",
|
"description": "Tools for making a Nostr client.",
|
||||||
"repository": {
|
"repository": {
|
||||||
"type": "git",
|
"type": "git",
|
||||||
|
|||||||
122
pool.ts
122
pool.ts
@@ -7,26 +7,38 @@ import {
|
|||||||
import {normalizeURL} from './utils.ts'
|
import {normalizeURL} from './utils.ts'
|
||||||
|
|
||||||
import type {Event} from './event.ts'
|
import type {Event} from './event.ts'
|
||||||
import type {Filter} from './filter.ts'
|
import {matchFilters, type Filter} from './filter.ts'
|
||||||
|
|
||||||
|
type BatchedRequest = {
|
||||||
|
filters: Filter<any>[]
|
||||||
|
relays: string[]
|
||||||
|
resolve: (events: Event<any>[]) => void
|
||||||
|
events: Event<any>[]
|
||||||
|
}
|
||||||
|
|
||||||
export class SimplePool {
|
export class SimplePool {
|
||||||
private _conn: {[url: string]: Relay}
|
private _conn: {[url: string]: Relay}
|
||||||
private _seenOn: {[id: string]: Set<string>} = {} // a map of all events we've seen in each 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 eoseSubTimeout: number
|
||||||
private getTimeout: number
|
private getTimeout: number
|
||||||
private seenOnEnabled: boolean = true
|
private seenOnEnabled: boolean = true
|
||||||
|
private batchInterval: number = 100
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
options: {
|
options: {
|
||||||
eoseSubTimeout?: number
|
eoseSubTimeout?: number
|
||||||
getTimeout?: number
|
getTimeout?: number
|
||||||
seenOnEnabled?: boolean
|
seenOnEnabled?: boolean
|
||||||
|
batchInterval?: number
|
||||||
} = {}
|
} = {}
|
||||||
) {
|
) {
|
||||||
this._conn = {}
|
this._conn = {}
|
||||||
this.eoseSubTimeout = options.eoseSubTimeout || 3400
|
this.eoseSubTimeout = options.eoseSubTimeout || 3400
|
||||||
this.getTimeout = options.getTimeout || 3400
|
this.getTimeout = options.getTimeout || 3400
|
||||||
this.seenOnEnabled = options.seenOnEnabled !== false
|
this.seenOnEnabled = options.seenOnEnabled !== false
|
||||||
|
this.batchInterval = options.batchInterval || 100
|
||||||
}
|
}
|
||||||
|
|
||||||
close(relays: string[]): void {
|
close(relays: string[]): void {
|
||||||
@@ -81,34 +93,36 @@ export class SimplePool {
|
|||||||
for (let cb of eoseListeners.values()) cb()
|
for (let cb of eoseListeners.values()) cb()
|
||||||
}, this.eoseSubTimeout)
|
}, this.eoseSubTimeout)
|
||||||
|
|
||||||
relays.forEach(async relay => {
|
relays
|
||||||
let r
|
.filter((r, i, a) => a.indexOf(r) == i)
|
||||||
try {
|
.forEach(async relay => {
|
||||||
r = await this.ensureRelay(relay)
|
let r
|
||||||
} catch (err) {
|
try {
|
||||||
handleEose()
|
r = await this.ensureRelay(relay)
|
||||||
return
|
} catch (err) {
|
||||||
}
|
handleEose()
|
||||||
if (!r) return
|
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() {
|
|
||||||
eosesMissing--
|
|
||||||
if (eosesMissing === 0) {
|
|
||||||
clearTimeout(eoseTimeout)
|
|
||||||
for (let cb of eoseListeners.values()) cb()
|
|
||||||
}
|
}
|
||||||
}
|
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() {
|
||||||
|
eosesMissing--
|
||||||
|
if (eosesMissing === 0) {
|
||||||
|
clearTimeout(eoseTimeout)
|
||||||
|
for (let cb of eoseListeners.values()) cb()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
let greaterSub: Sub = {
|
let greaterSub: Sub = {
|
||||||
sub(filters, opts) {
|
sub(filters, opts) {
|
||||||
@@ -176,6 +190,58 @@ export class SimplePool {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
batchedList<K extends number = number>(
|
||||||
|
batchKey: string,
|
||||||
|
relays: string[],
|
||||||
|
filters: Filter<K>[]
|
||||||
|
): Promise<Event<K>[]> {
|
||||||
|
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, 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<number>): Promise<void>[] {
|
publish(relays: string[], event: Event<number>): Promise<void>[] {
|
||||||
return relays.map(async relay => {
|
return relays.map(async relay => {
|
||||||
let r = await this.ensureRelay(relay)
|
let r = await this.ensureRelay(relay)
|
||||||
|
|||||||
Reference in New Issue
Block a user