From d1f2b6bd4f5f48ac59129877d7e24663d280b21e Mon Sep 17 00:00:00 2001 From: Jon Staab Date: Tue, 25 Mar 2025 09:59:11 -0700 Subject: [PATCH] Distinguish between unicast and multicast --- packages/net2/src/diff.ts | 15 +--- packages/net2/src/publish.ts | 152 ++++++++++++++++++++++++++--------- 2 files changed, 117 insertions(+), 50 deletions(-) diff --git a/packages/net2/src/diff.ts b/packages/net2/src/diff.ts index 62bb663..4300cbf 100644 --- a/packages/net2/src/diff.ts +++ b/packages/net2/src/diff.ts @@ -12,7 +12,7 @@ import { import {getAdapter, AdapterContext, AbstractAdapter, AdapterEventType} from "./adapter.js" import {Negentropy, NegentropyStorageVector} from "./negentropy.js" import {subscribe, SubscriptionEventType} from "./subscribe.js" -import {publish, PublicationEventType} from "./publish.js" +import {publish, PublishEventType} from "./publish.js" export enum DifferenceEventType { Message = "difference:event:message", @@ -235,16 +235,9 @@ export const push = async ({context, events, ...options}: PushOptions) => { const relays = relaysById.get(event.id) if (relays) { - await Promise.all( - relays.map( - relay => - new Promise(resolve => { - const pub = publish({event, relay, context}) - - pub.on(PublicationEventType.Complete, resolve) - }), - ), - ) + new Promise(resolve => { + publish({event, relays, context}).on(PublishEventType.Complete, resolve) + }) } }), ) diff --git a/packages/net2/src/publish.ts b/packages/net2/src/publish.ts index bb7a792..0cf5365 100644 --- a/packages/net2/src/publish.ts +++ b/packages/net2/src/publish.ts @@ -1,55 +1,54 @@ import {EventEmitter} from "events" -import {on, sleep, yieldThread} from "@welshman/lib" +import {on, fromPairs, sleep, yieldThread} from "@welshman/lib" import {SignedEvent} from "@welshman/util" import {RelayMessage, ClientMessageType, isRelayOk} from "./message.js" import {AbstractAdapter, AdapterEventType, AdapterContext, getAdapter} from "./adapter.js" import {TypedEmitter} from "./util.js" -export enum PublicationStatus { - Pending = "publication:status:pending", - Success = "publication:status:success", - Failure = "publication:status:failure", - Timeout = "publication:status:timeout", - Aborted = "publication:status:aborted", +export enum PublishStatus { + Pending = "publish:status:pending", + Success = "publish:status:success", + Failure = "publish:status:failure", + Timeout = "publish:status:timeout", + Aborted = "publish:status:aborted", } -export enum PublicationEventType { - Success = "publication:event:success", - Failure = "publication:event:failure", - Timeout = "publication:event:timeout", - Aborted = "publication:event:aborted", - Complete = "publication:event:complete", +export enum PublishEventType { + Success = "publish:event:success", + Failure = "publish:event:failure", + Timeout = "publish:event:timeout", + Aborted = "publish:event:aborted", + Complete = "publish:event:complete", } -export type PublicationEvents = { - [PublicationEventType.Success]: (id: string, detail: string, url: string) => void - [PublicationEventType.Failure]: (id: string, detail: string, url: string) => void - [PublicationEventType.Timeout]: () => void - [PublicationEventType.Aborted]: () => void - [PublicationEventType.Complete]: () => void +export type UnicastEvents = { + [PublishEventType.Success]: (id: string, detail: string) => void + [PublishEventType.Failure]: (id: string, detail: string) => void + [PublishEventType.Timeout]: () => void + [PublishEventType.Aborted]: () => void + [PublishEventType.Complete]: () => void } -export type PublicationOptions = { - relay: string +export type UnicastOptions = { event: SignedEvent + relay: string context: AdapterContext timeout?: number } -export class Publication extends (EventEmitter as new () => TypedEmitter) { - status = PublicationStatus.Pending +export class Unicast extends (EventEmitter as new () => TypedEmitter) { + status = PublishStatus.Pending - _done = new Set() _unsubscriber: () => void _adapter: AbstractAdapter - constructor(readonly options: PublicationOptions) { + constructor(readonly options: UnicastOptions) { super() // Set up our adapter this._adapter = getAdapter(this.options.relay, this.options.context) - // Listen for Publication result + // Listen for Unicast result this._unsubscriber = on( this._adapter, AdapterEventType.Receive, @@ -60,11 +59,11 @@ export class Publication extends (EventEmitter as new () => TypedEmitter TypedEmitter { // Set timeout sleep(this.options.timeout || 10_000).then(() => { - if (this.status === PublicationStatus.Pending) { - this.status = PublicationStatus.Timeout - this.emit(PublicationEventType.Timeout) + if (this.status === PublishStatus.Pending) { + this.status = PublishStatus.Timeout + this.emit(PublishEventType.Timeout) } this.cleanup() }) - // Send the Publication message + // Send the message this._adapter.send([ClientMessageType.Event, event]) } abort = () => { - if (this.status === PublicationStatus.Pending) { - this.status = PublicationStatus.Aborted - this.emit(PublicationEventType.Aborted) + if (this.status === PublishStatus.Pending) { + this.status = PublishStatus.Aborted + this.emit(PublishEventType.Aborted) this.cleanup() } } cleanup = () => { - this.emit(PublicationEventType.Complete) + this.emit(PublishEventType.Complete) this.removeAllListeners() this._adapter.cleanup() this._unsubscriber() } } -export const publish = (options: PublicationOptions) => new Publication(options) +export type MulticastEvents = { + [PublishEventType.Success]: (id: string, detail: string, url: string) => void + [PublishEventType.Failure]: (id: string, detail: string, url: string) => void + [PublishEventType.Timeout]: (url: string) => void + [PublishEventType.Aborted]: (url: string) => void + [PublishEventType.Complete]: () => void +} + +export type MulticastOptions = { + event: SignedEvent + relays: string[] + context: AdapterContext + timeout?: number +} + +export class Multicast extends (EventEmitter as new () => TypedEmitter) { + status: Record + + _children: Unicast[] = [] + _completed = new Set() + + constructor({relays, ...options}: MulticastOptions) { + super() + + this.status = fromPairs(relays.map(relay => [relay, PublishStatus.Pending])) + + for (const relay of relays) { + const unicast = new Unicast({relay, ...options}) + + unicast.on(PublishEventType.Success, (id: string, detail: string) => { + this.status[relay] = unicast.status + this.emit(PublishEventType.Success, id, detail, relay) + }) + + unicast.on(PublishEventType.Failure, (id: string, detail: string) => { + this.status[relay] = unicast.status + this.emit(PublishEventType.Failure, id, detail, relay) + }) + + unicast.on(PublishEventType.Timeout, () => { + this.status[relay] = unicast.status + this.emit(PublishEventType.Timeout, relay) + }) + + unicast.on(PublishEventType.Aborted, () => { + this.status[relay] = unicast.status + this.emit(PublishEventType.Aborted, relay) + }) + + unicast.on(PublishEventType.Complete, () => { + this._completed.add(relay) + this.status[relay] = unicast.status + + if (this._completed.size === relays.length) { + this.emit(PublishEventType.Complete) + } + }) + + this._children.push(unicast) + } + } + + abort() { + for (const child of this._children) { + child.abort() + } + } + + cleanup() { + for (const child of this._children) { + child.cleanup() + } + } +} + +export const publish = (options: MulticastOptions) => new Multicast(options)