Read remote RTS publications through CultMesh documents
This commit is contained in:
parent
ed2e547aeb
commit
294a296ccd
2 changed files with 29 additions and 74 deletions
|
|
@ -1,10 +1,5 @@
|
|||
import { decode } from "@msgpack/msgpack";
|
||||
import { CultMesh } from "cultmesh-ts";
|
||||
import type { CultNetPeer } from "cultnet-ts";
|
||||
import type {
|
||||
CultNetErrorMessage,
|
||||
CultNetSnapshotResponseRawMessage,
|
||||
} from "cultnet-ts";
|
||||
|
||||
const connectionId = 0x43554c54;
|
||||
|
||||
|
|
@ -51,67 +46,17 @@ export class AetheriaRemotePublicationReader {
|
|||
}
|
||||
|
||||
private async readSingleDocument(schemaId: string, recordKey: string): Promise<unknown> {
|
||||
const response = await this.requestSnapshot([recordKey]);
|
||||
const document = response.documents.find(candidate =>
|
||||
candidate.recordKey === recordKey);
|
||||
if (!document) {
|
||||
throw new Error(`Aetheria remote publication ${this.endpoint} did not return ${schemaId} at ${recordKey}.`);
|
||||
}
|
||||
|
||||
return decode(toBytes(document.payload));
|
||||
}
|
||||
|
||||
private async requestSnapshot(
|
||||
recordKeys: readonly string[],
|
||||
): Promise<CultNetSnapshotResponseRawMessage> {
|
||||
const peer = await this.peer();
|
||||
const messageId = `${this.runtimeId}:snapshot:${Date.now()}:${Math.random().toString(16).slice(2)}`;
|
||||
|
||||
return new Promise<CultNetSnapshotResponseRawMessage>((resolve, reject) => {
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timer);
|
||||
peer.off("message", onMessage);
|
||||
peer.off("invalidMessage", onInvalidMessage);
|
||||
peer.off("error", onError);
|
||||
peer.off("close", onClose);
|
||||
};
|
||||
const rejectWith = (error: Error): void => {
|
||||
cleanup();
|
||||
reject(error);
|
||||
};
|
||||
const onMessage = (message: unknown): void => {
|
||||
const response = message as CultNetSnapshotResponseRawMessage | CultNetErrorMessage;
|
||||
if (response.schemaVersion === "cultnet.error.v0") {
|
||||
cleanup();
|
||||
reject(new Error(response.error));
|
||||
return;
|
||||
}
|
||||
|
||||
if (response.schemaVersion !== "cultnet.snapshot_response_raw.v0" ||
|
||||
response.messageId !== messageId) {
|
||||
return;
|
||||
}
|
||||
|
||||
cleanup();
|
||||
resolve(response);
|
||||
};
|
||||
const onInvalidMessage = (error: Error): void => rejectWith(error);
|
||||
const onError = (error: Error): void => rejectWith(error);
|
||||
const onClose = (): void => rejectWith(new Error("CultMesh peer closed before snapshot response."));
|
||||
const timer = setTimeout(
|
||||
() => rejectWith(new Error(`Timed out waiting for Aetheria remote snapshot ${messageId}.`)),
|
||||
this.timeoutMs);
|
||||
|
||||
peer.on("message", onMessage);
|
||||
peer.on("invalidMessage", onInvalidMessage);
|
||||
peer.on("error", onError);
|
||||
peer.on("close", onClose);
|
||||
peer.sendSnapshotRequest({
|
||||
schemaVersion: "cultnet.snapshot_request.v0",
|
||||
messageId,
|
||||
recordKeys: [...recordKeys],
|
||||
});
|
||||
});
|
||||
return CultMesh.documentFromPeerSnapshot(
|
||||
() => this.peer(),
|
||||
schemaId,
|
||||
recordKey,
|
||||
{
|
||||
documentId: recordKey,
|
||||
routeHint: CultMesh.routeHint("network", this.endpoint),
|
||||
timeoutMs: this.timeoutMs,
|
||||
messageIdPrefix: `${this.runtimeId}:snapshot`,
|
||||
},
|
||||
).latest();
|
||||
}
|
||||
|
||||
private async peer(): Promise<CultNetPeer> {
|
||||
|
|
@ -129,11 +74,3 @@ export class AetheriaRemotePublicationReader {
|
|||
return this.#peer;
|
||||
}
|
||||
}
|
||||
|
||||
function toBytes(value: unknown): Uint8Array {
|
||||
if (value instanceof Uint8Array)
|
||||
return value;
|
||||
if (Array.isArray(value))
|
||||
return Uint8Array.from(value);
|
||||
throw new Error("CultNet raw document payload was not binary.");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -102,6 +102,24 @@ if ($LASTEXITCODE -gt 1) {
|
|||
Write-Error "Stage 7B verifier could not run rg for transport wrapper layout checks."
|
||||
}
|
||||
|
||||
$remotePublicationReaderPatterns = @(
|
||||
'@msgpack/msgpack',
|
||||
'sendSnapshotRequest',
|
||||
'snapshot_response_raw',
|
||||
'CultNetSnapshotResponseRawMessage',
|
||||
'recordKeys',
|
||||
'decode\('
|
||||
)
|
||||
|
||||
$remotePublicationReaderPattern = ($remotePublicationReaderPatterns -join '|')
|
||||
$remotePublicationReaderHits = & rg -n $remotePublicationReaderPattern Electron/aetheria-remote-publication-reader.ts 2>$null
|
||||
if ($LASTEXITCODE -eq 0) {
|
||||
Write-Error "Stage 7B verifier failed: RTS remote publication reader still owns raw snapshot/codec details.`n$remotePublicationReaderHits"
|
||||
}
|
||||
if ($LASTEXITCODE -gt 1) {
|
||||
Write-Error "Stage 7B verifier could not run rg for remote publication reader layout checks."
|
||||
}
|
||||
|
||||
$requiredBindingSymbols = @(
|
||||
'createAetheriaRuntimeRtsOperationHandles',
|
||||
'createAetheriaRuntimeRtsQueryHandles',
|
||||
|
|
|
|||
Loading…
Reference in a new issue