Compare commits
3 Commits
282b41d2a6
...
6ca8e9161a
Author | SHA1 | Date |
---|---|---|
|
6ca8e9161a | |
|
c9ff51e278 | |
|
57f297be52 |
|
@ -78,14 +78,14 @@ export class AbstractSimplePool {
|
|||
for (let i = 0; i < relays.length; i++) {
|
||||
const url = normalizeURL(relays[i])
|
||||
if (!request.find(r => r.url === url)) {
|
||||
request.push({ url, filter })
|
||||
request.push({ url, filter: filter })
|
||||
}
|
||||
}
|
||||
|
||||
return this.subscribeMap(request, params)
|
||||
}
|
||||
|
||||
subscribeMany(relays: string[], filters: Filter[], params: SubscribeManyParams): SubCloser {
|
||||
subscribeMany(relays: string[], filter: Filter, params: SubscribeManyParams): SubCloser {
|
||||
params.onauth = params.onauth || params.doauth
|
||||
|
||||
const request: { url: string; filter: Filter }[] = []
|
||||
|
@ -93,9 +93,8 @@ export class AbstractSimplePool {
|
|||
for (let i = 0; i < relays.length; i++) {
|
||||
const url = normalizeURL(relays[i])
|
||||
if (uniqUrls.indexOf(url) === -1) {
|
||||
for (let f = 0; f < filters.length; f++) {
|
||||
request.push({ url, filter: filters[f] })
|
||||
}
|
||||
uniqUrls.push(url)
|
||||
request.push({ url, filter: filter })
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -105,6 +104,14 @@ export class AbstractSimplePool {
|
|||
subscribeMap(requests: { url: string; filter: Filter }[], params: SubscribeManyParams): SubCloser {
|
||||
params.onauth = params.onauth || params.doauth
|
||||
|
||||
const grouped = new Map<string, Filter[]>()
|
||||
for (const req of requests) {
|
||||
const { url, filter } = req
|
||||
if (!grouped.has(url)) grouped.set(url, [])
|
||||
grouped.get(url)!.push(filter)
|
||||
}
|
||||
const groupedRequests = Array.from(grouped.entries()).map(([url, filters]) => ({ url, filters }))
|
||||
|
||||
if (this.trackRelays) {
|
||||
params.receivedEvent = (relay: AbstractRelay, id: string) => {
|
||||
let set = this.seenOn.get(id)
|
||||
|
@ -152,7 +159,7 @@ export class AbstractSimplePool {
|
|||
|
||||
// open a subscription in all given relays
|
||||
const allOpened = Promise.all(
|
||||
requests.map(async ({ url, filter }, i) => {
|
||||
groupedRequests.map(async ({ url, filters }, i) => {
|
||||
let relay: AbstractRelay
|
||||
try {
|
||||
relay = await this.ensureRelay(url, {
|
||||
|
@ -163,7 +170,7 @@ export class AbstractSimplePool {
|
|||
return
|
||||
}
|
||||
|
||||
let subscription = relay.subscribe([filter], {
|
||||
let subscription = relay.subscribe(filters, {
|
||||
...params,
|
||||
oneose: () => handleEose(i),
|
||||
onclose: reason => {
|
||||
|
@ -171,7 +178,7 @@ export class AbstractSimplePool {
|
|||
relay
|
||||
.auth(params.onauth)
|
||||
.then(() => {
|
||||
relay.subscribe([filter], {
|
||||
relay.subscribe(filters, {
|
||||
...params,
|
||||
oneose: () => handleEose(i),
|
||||
onclose: reason => {
|
||||
|
@ -224,12 +231,12 @@ export class AbstractSimplePool {
|
|||
|
||||
subscribeManyEose(
|
||||
relays: string[],
|
||||
filters: Filter[],
|
||||
filter: Filter,
|
||||
params: Pick<SubscribeManyParams, 'label' | 'id' | 'onevent' | 'onclose' | 'maxWait' | 'onauth' | 'doauth'>,
|
||||
): SubCloser {
|
||||
params.onauth = params.onauth || params.doauth
|
||||
|
||||
const subcloser = this.subscribeMany(relays, filters, {
|
||||
const subcloser = this.subscribeMany(relays, filter, {
|
||||
...params,
|
||||
oneose() {
|
||||
subcloser.close('closed automatically on eose')
|
||||
|
|
|
@ -426,7 +426,7 @@ export class Subscription {
|
|||
}
|
||||
|
||||
public fire() {
|
||||
this.relay.send('["REQ","' + this.id + '",' + JSON.stringify(this.filters).substring(1))
|
||||
this.relay.send(JSON.stringify(['REQ', this.id, ...this.filters]))
|
||||
|
||||
// only now we start counting the eoseTimeout
|
||||
this.eoseTimeoutHandle = setTimeout(this.receivedEose.bind(this), this.eoseTimeout)
|
||||
|
|
24
pool.test.ts
24
pool.test.ts
|
@ -35,14 +35,18 @@ test('removing duplicates when subscribing', async () => {
|
|||
priv,
|
||||
)
|
||||
|
||||
pool.subscribeMany(relayURLs, [{ authors: [pub] }], {
|
||||
onevent(event: Event) {
|
||||
// this should be called only once even though we're listening
|
||||
// to multiple relays because the events will be caught and
|
||||
// deduplicated efficiently (without even being parsed)
|
||||
received.push(event)
|
||||
pool.subscribeMany(
|
||||
relayURLs,
|
||||
{ authors: [pub] },
|
||||
{
|
||||
onevent(event: Event) {
|
||||
// this should be called only once even though we're listening
|
||||
// to multiple relays because the events will be caught and
|
||||
// deduplicated efficiently (without even being parsed)
|
||||
received.push(event)
|
||||
},
|
||||
},
|
||||
})
|
||||
)
|
||||
|
||||
await Promise.any(pool.publish(relayURLs, event))
|
||||
await new Promise(resolve => setTimeout(resolve, 200)) // wait for the new published event to be received
|
||||
|
@ -55,12 +59,12 @@ test('same with double subs', async () => {
|
|||
let priv = generateSecretKey()
|
||||
let pub = getPublicKey(priv)
|
||||
|
||||
pool.subscribeMany(relayURLs, [{ authors: [pub] }], {
|
||||
pool.subscribeMany(relayURLs, { authors: [pub] }, {
|
||||
onevent(event) {
|
||||
received.push(event)
|
||||
},
|
||||
})
|
||||
pool.subscribeMany(relayURLs, [{ authors: [pub] }], {
|
||||
pool.subscribeMany(relayURLs, { authors: [pub] }, {
|
||||
onevent(event) {
|
||||
received.push(event)
|
||||
},
|
||||
|
@ -168,7 +172,7 @@ test('query a bunch of events and cancel on eose', async () => {
|
|||
let events = new Set<string>()
|
||||
|
||||
await new Promise<void>(resolve => {
|
||||
pool.subscribeManyEose(relayURLs, [{ kinds: [0, 1, 2, 3, 4, 5, 6], limit: 40 }], {
|
||||
pool.subscribeManyEose(relayURLs, { kinds: [0, 1, 2, 3, 4, 5, 6], limit: 40 }, {
|
||||
onevent(event) {
|
||||
events.add(event.id)
|
||||
},
|
||||
|
|
Loading…
Reference in New Issue