moon
/
ditto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reqmeister: accept `signal`

This commit is contained in:
Alex Gleason 2023-12-28 13:41:04 -06:00
parent 2d19ab207e
commit 08da26b12a
No known key found for this signature in database
GPG Key ID: 7211D1F99744FBB7
3 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipeline.ts
View File

@ -144,7 +144,7 @@ function fetchRelatedEvents(event: Event, data: EventData) {
} }
for (const [name, id, relay] of event.tags) { for (const [name, id, relay] of event.tags) {
if (name === 'e' && !memorelay.hasEventById(id)) { if (name === 'e' && !memorelay.hasEventById(id)) {
reqmeister.req({ ids: [id] }, [relay]).catch(() => {}); reqmeister.req({ ids: [id] }, { relays: [relay] }).catch(() => {});
} }
} }
} }

4
src/queries.ts
View File

@ -52,7 +52,7 @@ const getEvent = async <K extends number = number>(
if (memoryEvent) return memoryEvent; if (memoryEvent) return memoryEvent;
return await reqmeister.req(microfilter).catch(() => undefined) as Event<K> | undefined; return await reqmeister.req(microfilter, opts).catch(() => undefined) as Event<K> | undefined;
}; };
/** Get a Nostr `set_medatadata` event for a user's pubkey. */ /** Get a Nostr `set_medatadata` event for a user's pubkey. */
@ -74,7 +74,7 @@ const getAuthor = async (pubkey: string, opts: GetEventOpts<0> = {}): Promise<Ev
if (dbEvent) return dbEvent; if (dbEvent) return dbEvent;
if (memoryEvent) return memoryEvent; if (memoryEvent) return memoryEvent;
return reqmeister.req(microfilter).catch(() => undefined); return reqmeister.req(microfilter, opts).catch(() => undefined);
}; };
/** Get users the given pubkey follows. */ /** Get users the given pubkey follows. */

19
src/reqmeister.ts
View File

@ -10,6 +10,11 @@ interface ReqmeisterOpts {
timeout?: number; timeout?: number;
} }
interface ReqmeisterReqOpts {
relays?: WebSocket['url'][];
signal?: AbortSignal;
}
type ReqmeisterQueueItem = [string, MicroFilter, WebSocket['url'][]]; type ReqmeisterQueueItem = [string, MicroFilter, WebSocket['url'][]];
/** Batches requests to Nostr relays using microfilters. */ /** Batches requests to Nostr relays using microfilters. */
@ -70,15 +75,21 @@ class Reqmeister extends EventEmitter<{ [filterId: string]: (event: Event) => an
this.#perform(); this.#perform();
} }
req(filter: IdMicrofilter, relays?: WebSocket['url'][]): Promise<Event>; req(filter: IdMicrofilter, opts?: ReqmeisterReqOpts): Promise<Event>;
req(filter: AuthorMicrofilter, relays?: WebSocket['url'][]): Promise<Event<0>>; req(filter: AuthorMicrofilter, opts?: ReqmeisterReqOpts): Promise<Event<0>>;
req(filter: MicroFilter, relays?: WebSocket['url'][]): Promise<Event>; req(filter: MicroFilter, opts?: ReqmeisterReqOpts): Promise<Event>;
req(filter: MicroFilter, relays: WebSocket['url'][] = []): Promise<Event> { req(filter: MicroFilter, opts: ReqmeisterReqOpts = {}): Promise<Event> {
const { relays = [], signal } = opts;
if (signal?.aborted) return Promise.reject(new DOMException('Aborted', 'AbortError'));
const filterId = getFilterId(filter); const filterId = getFilterId(filter);
this.#queue.push([filterId, filter, relays]); this.#queue.push([filterId, filter, relays]);
return new Promise<Event>((resolve, reject) => { return new Promise<Event>((resolve, reject) => {
this.once(filterId, resolve); this.once(filterId, resolve);
this.#promise.finally(() => setTimeout(reject, 0)); this.#promise.finally(() => setTimeout(reject, 0));
signal?.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true });
}); });
} }