Add multireq/unireq

packages/net2/src/diff.ts

@@ -11,8 +11,8 @@ import {
 } from "./message.js"
 import {getAdapter, AdapterContext, AbstractAdapter, AdapterEventType} from "./adapter.js"
 import {Negentropy, NegentropyStorageVector} from "./negentropy.js"
-import {subscribe, SubscriptionEventType} from "./subscribe.js"
+import {unireq, RequestEventType} from "./request.js"
-import {publish, PublishEventType} from "./publish.js"
+import {multicast, PublishEventType} from "./publish.js"
 
 export enum DifferenceEventType {
   Message = "difference:event:message",
@@ -204,10 +204,10 @@ export const pull = async ({context, ...options}: PullOptions) => {
       return Promise.all(
         chunk(500, allIds).map(ids => {
           return new Promise<void>(resolve => {
-            const sub = subscribe({relay, context, filter: {ids}, autoClose: true})
+            const req = unireq({relay, context, filter: {ids}, autoClose: true})
 
-            sub.on(SubscriptionEventType.Close, resolve)
+            req.on(RequestEventType.Close, resolve)
-            sub.on(SubscriptionEventType.Event, event => result.push(event))
+            req.on(RequestEventType.Event, event => result.push(event))
           })
         }),
       )
@@ -236,7 +236,7 @@ export const push = async ({context, events, ...options}: PushOptions) => {
 
       if (relays) {
         new Promise<void>(resolve => {
-          publish({event, relays, context}).on(PublishEventType.Complete, resolve)
+          multicast({event, relays, context}).on(PublishEventType.Complete, resolve)
         })
       }
     }),

packages/net2/src/index.ts

@@ -7,5 +7,5 @@ export * from "./policy.js"
 export * from "./pool.js"
 export * from "./publish.js"
 export * from "./socket.js"
-export * from "./subscribe.js"
+export * from "./request.js"
 export * from "./tracker.js"

packages/net2/src/publish.ts

@@ -21,6 +21,8 @@ export enum PublishEventType {
   Complete = "publish:event:complete",
 }
 
+// Unicast
+
 export type UnicastEvents = {
   [PublishEventType.Success]: (id: string, detail: string) => void
   [PublishEventType.Failure]: (id: string, detail: string) => void
@@ -71,11 +73,6 @@ export class Unicast extends (EventEmitter as new () => TypedEmitter<UnicastEven
       },
     )
 
-    // Autostart asynchronously so the caller can set up listeners
-    yieldThread().then(this.start)
-  }
-
-  start = () => {
     // Set timeout
     sleep(this.options.timeout || 10_000).then(() => {
       if (this.status === PublishStatus.Pending) {
@@ -86,8 +83,10 @@ export class Unicast extends (EventEmitter as new () => TypedEmitter<UnicastEven
       this.cleanup()
     })
 
-    // Send the message
+    // Start asynchronously so the caller can set up listeners
-    this._adapter.send([ClientMessageType.Event, event])
+    yieldThread().then(() => {
+      this._adapter.send([ClientMessageType.Event, event])
+    })
   }
 
   abort = () => {
@@ -106,6 +105,8 @@ export class Unicast extends (EventEmitter as new () => TypedEmitter<UnicastEven
   }
 }
 
+// Multicast
+
 export type MulticastEvents = {
   [PublishEventType.Success]: (id: string, detail: string, url: string) => void
   [PublishEventType.Failure]: (id: string, detail: string, url: string) => void
@@ -114,11 +115,8 @@ export type MulticastEvents = {
   [PublishEventType.Complete]: () => void
 }
 
-export type MulticastOptions = {
+export type MulticastOptions = Omit<UnicastOptions, "relay"> & {
-  event: SignedEvent
   relays: string[]
-  context: AdapterContext
-  timeout?: number
 }
 
 export class Multicast extends (EventEmitter as new () => TypedEmitter<MulticastEvents>) {
@@ -181,4 +179,8 @@ export class Multicast extends (EventEmitter as new () => TypedEmitter<Multicast
   }
 }
 
-export const publish = (options: MulticastOptions) => new Multicast(options)
+// Convenience functions
+
+export const unicast = (options: UnicastOptions) => new Unicast(options)
+
+export const multicast = (options: MulticastOptions) => new Multicast(options)

packages/net2/src/request.ts

@@ -0,0 +1,198 @@
+import {EventEmitter} from "events"
+import {on, call, randomId, yieldThread} from "@welshman/lib"
+import {Filter, matchFilter, SignedEvent} from "@welshman/util"
+import {RelayMessage, ClientMessageType, isRelayEvent, isRelayEose} from "./message.js"
+import {getAdapter, AdapterContext, AbstractAdapter, AdapterEventType} from "./adapter.js"
+import {SocketEventType, SocketStatus} from "./socket.js"
+import {TypedEmitter, Unsubscriber} from "./util.js"
+import {Tracker} from "./tracker.js"
+
+export enum RequestEventType {
+  Close = "request:event:close",
+  Disconnect = "request:event:disconnect",
+  Duplicate = "request:event:duplicate",
+  Eose = "request:event:eose",
+  Event = "request:event:event",
+  Filtered = "request:event:filtered",
+  Invalid = "request:event:invalid",
+}
+
+// Unireq
+
+export type UnireqEvents = {
+  [RequestEventType.Event]: (event: SignedEvent) => void
+  [RequestEventType.Invalid]: (event: SignedEvent) => void
+  [RequestEventType.Filtered]: (event: SignedEvent) => void
+  [RequestEventType.Duplicate]: (event: SignedEvent) => void
+  [RequestEventType.Disconnect]: () => void
+  [RequestEventType.Close]: () => void
+  [RequestEventType.Eose]: () => void
+}
+
+export type UnireqOptions = {
+  relay: string
+  filter: Filter
+  context: AdapterContext
+  timeout?: number
+  tracker?: Tracker
+  autoClose?: boolean
+  verifyEvent?: (event: SignedEvent) => boolean
+}
+
+export class Unireq extends (EventEmitter as new () => TypedEmitter<UnireqEvents>) {
+  _id = `REQ-${randomId().slice(0, 8)}`
+  _unsubscribers: Unsubscriber[] = []
+  _adapter: AbstractAdapter
+  _closed = false
+
+  constructor(readonly options: UnireqOptions) {
+    super()
+
+    // Set up our adapter
+    this._adapter = getAdapter(this.options.relay, this.options.context)
+
+    // Listen for event/eose messages from the adapter
+    this._unsubscribers.push(
+      on(this._adapter, AdapterEventType.Receive, (message: RelayMessage, url: string) => {
+        if (isRelayEvent(message)) {
+          const [_, id, event] = message
+
+          if (id !== this._id) return
+
+          if (this.options.tracker?.track(event.id, url)) {
+            this.emit(RequestEventType.Duplicate, event)
+          } else if (this.options.verifyEvent?.(event) === false) {
+            this.emit(RequestEventType.Invalid, event)
+          } else if (!matchFilter(this.options.filter, event)) {
+            this.emit(RequestEventType.Filtered, event)
+          } else {
+            this.emit(RequestEventType.Event, event)
+          }
+        }
+
+        if (isRelayEose(message)) {
+          const [_, id] = message
+
+          if (id === this._id) {
+            this.emit(RequestEventType.Eose)
+
+            if (this.options.autoClose) {
+              this.close()
+            }
+          }
+        }
+      }),
+    )
+
+    // Listen to disconnects from any sockets
+    for (const socket of this._adapter.sockets) {
+      this._unsubscribers.push(
+        on(socket, SocketEventType.Status, (status: SocketStatus) => {
+          if (![SocketStatus.Open, SocketStatus.Opening].includes(status)) {
+            this.emit(RequestEventType.Disconnect)
+
+            if (this.options.autoClose) {
+              this.close()
+            }
+          }
+        }),
+      )
+    }
+
+    // Timeout our subscription
+    if (this.options.timeout) {
+      setTimeout(() => this.close(), this.options.timeout)
+    }
+
+    // Start asynchronously so the caller can set up listeners
+    yieldThread().then(() => {
+      this._adapter.send([ClientMessageType.Req, this._id, this.options.filter])
+    })
+  }
+
+  close() {
+    if (this._closed) return
+
+    this._adapter.send(["CLOSE", this._id])
+    this.emit(RequestEventType.Close)
+    this.removeAllListeners()
+    this._unsubscribers.map(call)
+    this._adapter.cleanup()
+    this._closed = true
+  }
+}
+
+// Multireq
+
+export type MultireqEvents = {
+  [RequestEventType.Event]: (event: SignedEvent, url: string) => void
+  [RequestEventType.Invalid]: (event: SignedEvent, url: string) => void
+  [RequestEventType.Filtered]: (event: SignedEvent, url: string) => void
+  [RequestEventType.Duplicate]: (event: SignedEvent, url: string) => void
+  [RequestEventType.Disconnect]: (url: string) => void
+  [RequestEventType.Eose]: (url: string) => void
+  [RequestEventType.Close]: () => void
+}
+
+export type MultireqOptions = Omit<UnireqOptions, "relay"> & {
+  relays: string[]
+}
+
+export class Multireq extends (EventEmitter as new () => TypedEmitter<MultireqEvents>) {
+  _children: Unireq[] = []
+  _closed = new Set<string>()
+
+  constructor({relays, ...options}: MultireqOptions) {
+    super()
+
+    for (const relay of relays) {
+      const req = new Unireq({relay, ...options})
+
+      req.on(RequestEventType.Event, (event: SignedEvent) => {
+        this.emit(RequestEventType.Event, event, relay)
+      })
+
+      req.on(RequestEventType.Invalid, (event: SignedEvent) => {
+        this.emit(RequestEventType.Invalid, event, relay)
+      })
+
+      req.on(RequestEventType.Filtered, (event: SignedEvent) => {
+        this.emit(RequestEventType.Filtered, event, relay)
+      })
+
+      req.on(RequestEventType.Duplicate, (event: SignedEvent) => {
+        this.emit(RequestEventType.Duplicate, event, relay)
+      })
+
+      req.on(RequestEventType.Disconnect, () => {
+        this.emit(RequestEventType.Disconnect, relay)
+      })
+
+      req.on(RequestEventType.Eose, () => {
+        this.emit(RequestEventType.Eose, relay)
+      })
+
+      req.on(RequestEventType.Close, () => {
+        this._closed.add(relay)
+
+        if (this._closed.size === relays.length) {
+          this.emit(RequestEventType.Close)
+        }
+      })
+
+      this._children.push(req)
+    }
+  }
+
+  close() {
+    for (const child of this._children) {
+      child.close()
+    }
+  }
+}
+
+// Convenience functions
+
+export const unireq = (options: UnireqOptions) => new Unireq(options)
+
+export const multireq = (options: MultireqOptions) => new Multireq(options)

packages/net2/src/subscribe.ts

@@ -1,126 +0,0 @@
-import {EventEmitter} from "events"
-import {on, call, randomId, yieldThread} from "@welshman/lib"
-import {Filter, matchFilter, SignedEvent} from "@welshman/util"
-import {RelayMessage, ClientMessageType, isRelayEvent, isRelayEose} from "./message.js"
-import {getAdapter, AdapterContext, AbstractAdapter, AdapterEventType} from "./adapter.js"
-import {SocketEventType, SocketStatus} from "./socket.js"
-import {TypedEmitter, Unsubscriber} from "./util.js"
-import {Tracker} from "./tracker.js"
-
-export enum SubscriptionEventType {
-  Close = "subscription:event:close",
-  Disconnect = "subscription:event:disconnect",
-  Duplicate = "subscription:event:duplicate",
-  Eose = "subscription:event:eose",
-  Event = "subscription:event:event",
-  Filtered = "subscription:event:filtered",
-  Invalid = "subscription:event:invalid",
-}
-
-export type SubscriptionEvents = {
-  [SubscriptionEventType.Close]: () => void
-  [SubscriptionEventType.Disconnect]: (url: string) => void
-  [SubscriptionEventType.Duplicate]: (event: SignedEvent, url: string) => void
-  [SubscriptionEventType.Eose]: (url: string) => void
-  [SubscriptionEventType.Event]: (event: SignedEvent, url: string) => void
-  [SubscriptionEventType.Filtered]: (event: SignedEvent, url: string) => void
-  [SubscriptionEventType.Invalid]: (event: SignedEvent, url: string) => void
-}
-
-export type SubscriptionOptions = {
-  relay: string
-  filter: Filter
-  context: AdapterContext
-  timeout?: number
-  tracker?: Tracker
-  autoClose?: boolean
-  verifyEvent?: (event: SignedEvent) => boolean
-}
-
-export class Subscription extends (EventEmitter as new () => TypedEmitter<SubscriptionEvents>) {
-  _id = `REQ-${randomId().slice(0, 8)}`
-  _unsubscribers: Unsubscriber[] = []
-  _adapter: AbstractAdapter
-  _closed = false
-
-  constructor(readonly options: SubscriptionOptions) {
-    super()
-
-    // Set up our adapter
-    this._adapter = getAdapter(this.options.relay, this.options.context)
-
-    // Listen for event/eose messages from the adapter
-    this._unsubscribers.push(
-      on(this._adapter, AdapterEventType.Receive, (message: RelayMessage, url: string) => {
-        if (isRelayEvent(message)) {
-          const [_, id, event] = message
-
-          if (id !== this._id) return
-
-          if (this.options.tracker?.track(event.id, url)) {
-            this.emit(SubscriptionEventType.Duplicate, event, url)
-          } else if (this.options.verifyEvent?.(event) === false) {
-            this.emit(SubscriptionEventType.Invalid, event, url)
-          } else if (!matchFilter(this.options.filter, event)) {
-            this.emit(SubscriptionEventType.Filtered, event, url)
-          } else {
-            this.emit(SubscriptionEventType.Event, event, url)
-          }
-        }
-
-        if (isRelayEose(message)) {
-          const [_, id] = message
-
-          if (id === this._id) {
-            this.emit(SubscriptionEventType.Eose, url)
-
-            if (this.options.autoClose) {
-              this.close()
-            }
-          }
-        }
-      }),
-    )
-
-    // Listen to disconnects from any sockets
-    for (const socket of this._adapter.sockets) {
-      this._unsubscribers.push(
-        on(socket, SocketEventType.Status, (status: SocketStatus) => {
-          if (![SocketStatus.Open, SocketStatus.Opening].includes(status)) {
-            this.emit(SubscriptionEventType.Disconnect, socket.url)
-
-            if (this.options.autoClose) {
-              this.close()
-            }
-          }
-        }),
-      )
-    }
-
-    // Autostart asynchronously so the caller can set up listeners
-    yieldThread().then(this.open)
-  }
-
-  open = () => {
-    // Timeout our subscription
-    if (this.options.timeout) {
-      setTimeout(() => this.close(), this.options.timeout)
-    }
-
-    // Send our request
-    this._adapter.send([ClientMessageType.Req, this._id, this.options.filter])
-  }
-
-  close() {
-    if (this._closed) return
-
-    this._adapter.send(["CLOSE", this._id])
-    this.emit(SubscriptionEventType.Close)
-    this.removeAllListeners()
-    this._unsubscribers.map(call)
-    this._adapter.cleanup()
-    this._closed = true
-  }
-}
-
-export const subscribe = (options: SubscriptionOptions) => new Subscription(options)