save wrap index, chunk filters in subscribe

This commit is contained in:
Jon Staab
2024-06-06 11:46:54 -07:00
parent 0b5f1589b2
commit 53fe1b0a51
2 changed files with 13 additions and 4 deletions
+8 -4
View File
@@ -1,5 +1,5 @@
import type {Event} from 'nostr-tools' import type {Event} from 'nostr-tools'
import {Emitter, flatten, randomId, once, groupBy, batch, defer, uniq, uniqBy} from '@welshman/lib' import {Emitter, chunk, flatten, randomId, once, groupBy, batch, defer, uniq, uniqBy} from '@welshman/lib'
import type {Deferred} from '@welshman/lib' import type {Deferred} from '@welshman/lib'
import {matchFilters, unionFilters} from '@welshman/util' import {matchFilters, unionFilters} from '@welshman/util'
import type {Filter} from '@welshman/util' import type {Filter} from '@welshman/util'
@@ -182,6 +182,7 @@ export const executeSubscription = (sub: Subscription) => {
const {result, request, emitter, tracker, controller} = sub const {result, request, emitter, tracker, controller} = sub
const {timeout, filters, closeOnEose, relays, signal} = request const {timeout, filters, closeOnEose, relays, signal} = request
const executor = NetworkContext.getExecutor(relays) const executor = NetworkContext.getExecutor(relays)
const subs: {unsubscribe: () => void}[] = []
const completedRelays = new Set() const completedRelays = new Set()
const events: Event[] = [] const events: Event[] = []
@@ -209,8 +210,8 @@ export const executeSubscription = (sub: Subscription) => {
emitter.on(SubscriptionEvent.Complete, () => { emitter.on(SubscriptionEvent.Complete, () => {
result.resolve(events) result.resolve(events)
executorSub?.unsubscribe()
emitter.removeAllListeners() emitter.removeAllListeners()
subs.forEach(sub => sub.unsubscribe())
executor.target.connections.forEach((c: Connection) => c.off("close", onClose)) executor.target.connections.forEach((c: Connection) => c.off("close", onClose))
executor.target.cleanup() executor.target.cleanup()
}) })
@@ -253,9 +254,12 @@ export const executeSubscription = (sub: Subscription) => {
// Finally, start our subscription. If we didn't get any filters, don't even send the // Finally, start our subscription. If we didn't get any filters, don't even send the
// request, just close it. This can be valid when a caller fulfills a request themselves. // request, just close it. This can be valid when a caller fulfills a request themselves.
let executorSub: {unsubscribe: () => void}
if (filters.length > 0) { if (filters.length > 0) {
executorSub = executor.subscribe(filters, {onEvent, onEose}) // If we send too many filters in a request relays will refuse to respond. REQs are rate
// limited client-side by Connection, so this will throttle concurrent requests.
for (const filtersChunk of chunk(8, filters)) {
subs.push(executor.subscribe(filtersChunk, {onEvent, onEose}))
}
} else { } else {
onComplete() onComplete()
} }
+5
View File
@@ -156,6 +156,11 @@ export class Repository extends Emitter {
this.eventsByAddress.set(address, event) this.eventsByAddress.set(address, event)
} }
// Save wrapper index
if (event.wrap) {
this.eventsByWrap.set(event.wrap.id, event)
}
// Update our timestamp and author indexes // Update our timestamp and author indexes
this._updateIndex(this.eventsByDay, getDay(event.created_at), event, duplicate) this._updateIndex(this.eventsByDay, getDay(event.created_at), event, duplicate)
this._updateIndex(this.eventsByAuthor, event.pubkey, event, duplicate) this._updateIndex(this.eventsByAuthor, event.pubkey, event, duplicate)