Distinguish between unicast and multicast

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(
+        new Promise<void>(resolve => {
-          relays.map(
+          publish({event, relays, context}).on(PublishEventType.Complete, resolve)
-            relay =>
+        })
-              new Promise<void>(resolve => {
-                const pub = publish({event, relay, context})
-
-                pub.on(PublicationEventType.Complete, resolve)
-              }),
-          ),
-        )
       }
     }),
   )

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 {
+export enum PublishStatus {
-  Pending = "publication:status:pending",
+  Pending = "publish:status:pending",
-  Success = "publication:status:success",
+  Success = "publish:status:success",
-  Failure = "publication:status:failure",
+  Failure = "publish:status:failure",
-  Timeout = "publication:status:timeout",
+  Timeout = "publish:status:timeout",
-  Aborted = "publication:status:aborted",
+  Aborted = "publish:status:aborted",
 }
 
-export enum PublicationEventType {
+export enum PublishEventType {
-  Success = "publication:event:success",
+  Success = "publish:event:success",
-  Failure = "publication:event:failure",
+  Failure = "publish:event:failure",
-  Timeout = "publication:event:timeout",
+  Timeout = "publish:event:timeout",
-  Aborted = "publication:event:aborted",
+  Aborted = "publish:event:aborted",
-  Complete = "publication:event:complete",
+  Complete = "publish:event:complete",
 }
 
-export type PublicationEvents = {
+export type UnicastEvents = {
-  [PublicationEventType.Success]: (id: string, detail: string, url: string) => void
+  [PublishEventType.Success]: (id: string, detail: string) => void
-  [PublicationEventType.Failure]: (id: string, detail: string, url: string) => void
+  [PublishEventType.Failure]: (id: string, detail: string) => void
-  [PublicationEventType.Timeout]: () => void
+  [PublishEventType.Timeout]: () => void
-  [PublicationEventType.Aborted]: () => void
+  [PublishEventType.Aborted]: () => void
-  [PublicationEventType.Complete]: () => void
+  [PublishEventType.Complete]: () => void
 }
 
-export type PublicationOptions = {
+export type UnicastOptions = {
-  relay: string
   event: SignedEvent
+  relay: string
   context: AdapterContext
   timeout?: number
 }
 
-export class Publication extends (EventEmitter as new () => TypedEmitter<PublicationEvents>) {
+export class Unicast extends (EventEmitter as new () => TypedEmitter<UnicastEvents>) {
-  status = PublicationStatus.Pending
+  status = PublishStatus.Pending
 
-  _done = new Set<string>()
   _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<Publica
           if (id !== this.options.event.id) return
 
           if (ok) {
-            this.status = PublicationStatus.Success
+            this.status = PublishStatus.Success
-            this.emit(PublicationEventType.Success, id, detail, url)
+            this.emit(PublishEventType.Success, id, detail)
           } else {
-            this.status = PublicationStatus.Failure
+            this.status = PublishStatus.Failure
-            this.emit(PublicationEventType.Failure, id, detail, url)
+            this.emit(PublishEventType.Failure, id, detail)
           }
 
           this.cleanup()
@@ -79,32 +78,107 @@ export class Publication extends (EventEmitter as new () => TypedEmitter<Publica
   start = () => {
     // Set timeout
     sleep(this.options.timeout || 10_000).then(() => {
-      if (this.status === PublicationStatus.Pending) {
+      if (this.status === PublishStatus.Pending) {
-        this.status = PublicationStatus.Timeout
+        this.status = PublishStatus.Timeout
-        this.emit(PublicationEventType.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) {
+    if (this.status === PublishStatus.Pending) {
-      this.status = PublicationStatus.Aborted
+      this.status = PublishStatus.Aborted
-      this.emit(PublicationEventType.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<MulticastEvents>) {
+  status: Record<string, PublishStatus>
+
+  _children: Unicast[] = []
+  _completed = new Set<string>()
+
+  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)