coracle/welshman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flesh out publish

Browse Source
This commit is contained in:
Jon Staab
2025-03-21 16:16:40 -07:00
parent 3df585698c
commit 44123426f2
5 changed files with 129 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/net2/src/adapter.ts
+1
View File
@@ -21,6 +21,7 @@ export abstract class AbstractAdapter extends (EventEmitter as new () => TypedEm
abstract send(message: ClientMessage): void
cleanup() {
this.removeAllListeners()
this._unsubscribers.splice(0).forEach(call)
}
}
packages/net2/src/diff.ts
+11 -5
View File
@@ -1,7 +1,13 @@
import {EventEmitter} from "events"
import {on, randomId} from "@welshman/lib"
import {SignedEvent, Filter} from "@welshman/util"
import {RelayMessage, isRelayNegErr, isRelayNegMsg} from "./message.js"
import {
RelayMessage,
isRelayNegErr,
isRelayNegMsg,
RelayMessageType,
ClientMessageType,
} from "./message.js"
import {AbstractAdapter, AdapterEventType} from "./adapter.js"
import {Negentropy, NegentropyStorageVector} from "./negentropy.js"
import {TypedEmitter} from "./util.js"
@@ -52,7 +58,7 @@ export class Diff extends (EventEmitter as new () => TypedEmitter<DiffEvents>) {
this.emit(DiffEventType.Message, {have, need}, url)
if (newMsg) {
adapter.send(["NEG-MSG", this._id, newMsg])
adapter.send([RelayMessageType.NegMsg, this._id, newMsg])
} else {
this.close()
}
@@ -70,17 +76,17 @@ export class Diff extends (EventEmitter as new () => TypedEmitter<DiffEvents>) {
)
neg.initiate().then((msg: string) => {
adapter.send(["NEG-OPEN", this._id, filter, msg])
adapter.send([ClientMessageType.NegOpen, this._id, filter, msg])
})
}
close() {
if (this._closed) return
this.adapter.send(["NEG-CLOSE", this._id])
this.adapter.send([ClientMessageType.NegClose, this._id])
this.emit(DiffEventType.Close)
this.removeAllListeners()
this._unsubscriber()
this._closed = true
}
}
packages/net2/src/publish.ts
+84 -15
View File
@@ -1,46 +1,115 @@
import {EventEmitter} from "events"
import {on} from "@welshman/lib"
import {on, sleep, yieldThread} from "@welshman/lib"
import {SignedEvent} from "@welshman/util"
import {RelayMessage, isRelayOk} from "./message.js"
import {RelayMessage, ClientMessageType, isRelayOk} from "./message.js"
import {AbstractAdapter, AdapterEventType} from "./adapter.js"
import {TypedEmitter} from "./util.js"
export enum PublishEventType {
Ok = "publish:event:ok",
export enum PublishStatus {
Pending = "publish:status:pending",
Success = "publish:status:success",
Failure = "publish:status:failure",
Timeout = "publish:status:timeout",
Aborted = "publish:status:aborted",
}
export type PublishEvents = {
[PublishEventType.Ok]: (id: string, ok: boolean, detail: string, url: string) => void
[PublishStatus.Pending]: (url: string) => void
[PublishStatus.Success]: (id: string, detail: string, url: string) => void
[PublishStatus.Failure]: (id: string, detail: string, url: string) => void
[PublishStatus.Timeout]: (url: string) => void
[PublishStatus.Aborted]: (url: string) => void
}
export type PublishOptions = {
adapter: AbstractAdapter
event: SignedEvent
timeout?: number
events?: Partial<PublishEvents>
}
export class Publish extends (EventEmitter as new () => TypedEmitter<PublishEvents>) {
status = new Map<string, PublishStatus>()
_done = new Set<string>()
_unsubscriber: () => void
constructor(
readonly adapter: AbstractAdapter,
readonly event: SignedEvent,
readonly verb = "EVENT",
) {
constructor(readonly options: PublishOptions) {
super()
// Listen for publish result
this._unsubscriber = on(
adapter,
this.options.adapter,
AdapterEventType.Receive,
(message: RelayMessage, url: string) => {
if (isRelayOk(message)) {
const [_, id, ok, detail] = message
if (id === event.id) {
this.emit(PublishEventType.Ok, id, ok, detail, url)
if (id !== this.options.event.id) return
if (ok) {
this.status.set(url, PublishStatus.Success)
this.emit(PublishStatus.Success, id, detail, url)
} else {
this.status.set(url, PublishStatus.Failure)
this.emit(PublishStatus.Failure, id, detail, url)
}
if (!Array.from(this.status.values()).includes(PublishStatus.Pending)) {
this.cleanup()
}
}
},
)
adapter.send([verb, event])
// Register handlers
if (this.options.events) {
for (const [k, listener] of Object.entries(this.options.events)) {
this.on(k as keyof PublishEvents, listener)
}
}
// Autostart asynchronously so the caller can set up listeners
yieldThread().then(this.start)
}
close() {
start = () => {
// Initialize status
for (const url of this.options.adapter.urls) {
this.status.set(url, PublishStatus.Pending)
this.emit(PublishStatus.Pending, url)
}
// Set timeout
sleep(this.options.timeout || 10_000).then(() => {
for (const [url, status] of this.status.entries()) {
if (status === PublishStatus.Pending) {
this.status.set(url, PublishStatus.Timeout)
this.emit(PublishStatus.Timeout, url)
}
}
this.cleanup()
})
// Send the publish message
this.options.adapter.send([ClientMessageType.Event, event])
}
abort = () => {
for (const [url, status] of this.status.entries()) {
if (status === PublishStatus.Pending) {
this.status.set(url, PublishStatus.Aborted)
this.emit(PublishStatus.Aborted, url)
}
}
this.cleanup()
}
cleanup = () => {
this.options.adapter.cleanup()
this.removeAllListeners()
this._unsubscriber()
}
}
packages/net2/src/socket.ts
+1
View File
@@ -109,6 +109,7 @@ export class Socket extends (EventEmitter as new () => TypedEmitter<SocketEvents
this.close()
this._recvQueue.clear()
this._sendQueue.clear()
this.removeAllListeners()
}
send = (message: ClientMessage) => {
packages/net2/src/subscribe.ts
+32 -20
View File
@@ -1,7 +1,7 @@
import {EventEmitter} from "events"
import {on, call, randomId} from "@welshman/lib"
import {on, call, randomId, yieldThread} from "@welshman/lib"
import {Filter, matchFilter, SignedEvent} from "@welshman/util"
import {RelayMessage, isRelayEvent, isRelayEose} from "./message.js"
import {RelayMessage, ClientMessageType, isRelayEvent, isRelayEose} from "./message.js"
import {AbstractAdapter, AdapterEventType} from "./adapter.js"
import {SocketEventType, SocketStatus} from "./socket.js"
import {TypedEmitter, Unsubscriber} from "./util.js"
@@ -40,26 +40,28 @@ export type SubscriptionOptions = {
export class Subscription extends (EventEmitter as new () => TypedEmitter<SubscriptionEvents>) {
_id = `REQ-${randomId().slice(0, 8)}`
_unsubscribers: Unsubscriber[] = []
_done = new Set<string>()
_closed = false
constructor(readonly options: SubscriptionOptions) {
super()
const done = new Set<string>()
const urls = new Set(options.adapter.urls)
// Get our unique urls so we know when we're done
const urls = new Set(this.options.adapter.urls)
// Listen for event/eose messages from the adapter
this._unsubscribers.push(
on(options.adapter, AdapterEventType.Receive, (message: RelayMessage, url: string) => {
on(this.options.adapter, AdapterEventType.Receive, (message: RelayMessage, url: string) => {
if (isRelayEvent(message)) {
const [_, id, event] = message
if (id !== this._id) return
if (options.tracker?.track(event.id, url)) {
if (this.options.tracker?.track(event.id, url)) {
this.emit(SubscriptionEventType.Duplicate, event, url)
} else if (options.verifyEvent?.(event) === false) {
} else if (this.options.verifyEvent?.(event) === false) {
this.emit(SubscriptionEventType.Invalid, event, url)
} else if (!matchFilter(options.filter, event)) {
} else if (!matchFilter(this.options.filter, event)) {
this.emit(SubscriptionEventType.Filtered, event, url)
} else {
this.emit(SubscriptionEventType.Event, event, url)
@@ -72,9 +74,9 @@ export class Subscription extends (EventEmitter as new () => TypedEmitter<Subscr
if (id === this._id) {
this.emit(SubscriptionEventType.Eose, url)
done.add(url)
this._done.add(url)
if (options.autoClose && done.size === urls.size) {
if (this.options.autoClose && this._done.size === urls.size) {
this.close()
}
}
@@ -82,15 +84,16 @@ export class Subscription extends (EventEmitter as new () => TypedEmitter<Subscr
}),
)
for (const socket of options.adapter.sockets) {
// Listen to disconnects from any sockets
for (const socket of this.options.adapter.sockets) {
this._unsubscribers.push(
on(socket, SocketEventType.Status, (status: SocketStatus) => {
if (![SocketStatus.Open, SocketStatus.Opening].includes(status)) {
this.emit(SubscriptionEventType.Disconnect, socket.url)
done.add(socket.url)
this._done.add(socket.url)
if (options.autoClose && done.size === urls.size) {
if (this.options.autoClose && this._done.size === urls.size) {
this.close()
}
}
@@ -98,17 +101,25 @@ export class Subscription extends (EventEmitter as new () => TypedEmitter<Subscr
)
}
if (options.timeout) {
setTimeout(() => this.close(), options.timeout)
}
if (options.events) {
for (const [k, listener] of Object.entries(options.events)) {
// Register listeners
if (this.options.events) {
for (const [k, listener] of Object.entries(this.options.events)) {
this.on(k as keyof SubscriptionEvents, listener)
}
}
options.adapter.send(["REQ", this._id, options.filter])
// 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.options.adapter.send([ClientMessageType.Req, this._id, this.options.filter])
}
close() {
@@ -116,6 +127,7 @@ export class Subscription extends (EventEmitter as new () => TypedEmitter<Subscr
this.options.adapter.send(["CLOSE", this._id])
this.emit(SubscriptionEventType.Close)
this.options.adapter.cleanup()
this.removeAllListeners()
this._unsubscribers.map(call)
this._closed = true