feat: simplify Chord service APIs
This commit is contained in:
@@ -278,7 +278,7 @@ A service has **one owner and many consumers**. In singleton mode, `providersBui
|
||||
- **Local:** `use()` returns a stable lazy proxy backed by a direct process-local implementation slot. During synchronous setup it is disconnected; after assembly it binds to the local implementation without requiring provider-before-consumer setup order. Reload unbinds and rebinds that same slot.
|
||||
- **Remote:** across a connection, `use()` returns the same kind of stable lazy proxy. Calls made while disconnected fail when invoked; state has no value until hydrated. Concurrent consumers of one token in one process share one proxy, one state replica, and one remote subscription.
|
||||
|
||||
Multi-instance services use `provideMany()` and `observe()`. The service is empty until its setup-owned `ServiceSpawner` calls `spawn()`; observing it never creates an instance. `spawner.spawn(key, implementation)` returns an idempotent close function, and the key must be unique among that service's live instances. Local observers use a direct process-local instance registry; non-local provisions additionally publish the same instance through RPC. `observe(service, handler)` reconciles current instances and then ordered additions, replacements, and removals. The handler receives the same `T` proxy shape as `use()`; `getServiceInstanceKey(proxy)` exposes its keyed application identity when needed. After an instance's initial state members hydrate, the host starts one handler task with a fresh `Context`. Closing the instance aborts that context, rejects new calls, and lets already-admitted calls return. Cancellation from the instance context is normal task cleanup; other handler failures follow host failure policy. Reusing a closed key creates a new host-owned generation, so stale proxies cannot address the replacement.
|
||||
Multi-instance services use `provideMany()` and `observe()`. The service is empty until its setup-owned `ServiceSpawner` calls `spawn()`; observing it never creates an instance. `spawner.spawn(key, implementation)` returns an idempotent close function, and the key must be unique among that service's live instances. Local observers use a direct process-local instance registry; non-local provisions additionally publish the same instance through RPC. `observe(service, handler)` reconciles current instances and then ordered additions, replacements, and removals. The handler receives the same `T` proxy shape as `use()`; instance keys remain provider-side addressing details. After an instance's initial state members hydrate, the host starts one handler task with a fresh `Context`. The facet lifecycle owns that observation. Closing the instance aborts its task context, rejects new calls, and lets already-admitted calls return. Cancellation from the instance context is normal task cleanup; other handler failures follow host failure policy. Reusing a closed key creates a new host-owned generation, so stale proxies cannot address the replacement.
|
||||
|
||||
An added instance member has structural identity `(service, key, generation, member)`. Its `ReplicatedState` members therefore need no independent IDs. The instance directory is control-plane metadata, not a facet-visible `ReplicatedState` containing proxies. Switching sessions aborts all observed instance tasks before hydrating the selected session's current instances.
|
||||
|
||||
@@ -319,7 +319,7 @@ interface FacetEnvironment extends FacetLifecycle {
|
||||
observe<T>(
|
||||
service: Service<T>,
|
||||
handler: (service: T, context: Context) => void | Promise<void>,
|
||||
): () => void;
|
||||
): void;
|
||||
provide<T>(service: Service<T>, implementation: T): void;
|
||||
provideMany<T>(service: Service<T>): ServiceSpawner<T>;
|
||||
replicatedState<T>(initial: T): MutableReplicatedState<T>;
|
||||
@@ -816,41 +816,39 @@ export const questionTuiFacet = defineFacet({
|
||||
id: "@pi/question",
|
||||
setup(env) {
|
||||
const tui = env.use(Tui);
|
||||
env.own(
|
||||
env.observe(QuestionDialogs, async (dialog, context) => {
|
||||
const request = dialog.service.request.value;
|
||||
if (request === undefined) throw new Error("Question dialog was observed before hydration");
|
||||
env.observe(QuestionDialogs, async (dialog, context) => {
|
||||
const request = dialog.request.value;
|
||||
if (request === undefined) throw new Error("Question dialog was observed before hydration");
|
||||
|
||||
const modal = await tui.acquireModal(context.abortSignal);
|
||||
try {
|
||||
const choice = await modal.select<QuestionChoice>(
|
||||
request.question,
|
||||
[
|
||||
...request.options.map((option, index) => ({
|
||||
label: option.label,
|
||||
...(option.description === null ? {} : { description: option.description }),
|
||||
value: { outcome: "selected" as const, index },
|
||||
})),
|
||||
{ label: "Write a custom answer", value: { outcome: "custom" as const } },
|
||||
],
|
||||
);
|
||||
const modal = await tui.acquireModal(context.abortSignal);
|
||||
try {
|
||||
const choice = await modal.select<QuestionChoice>(
|
||||
request.question,
|
||||
[
|
||||
...request.options.map((option, index) => ({
|
||||
label: option.label,
|
||||
...(option.description === null ? {} : { description: option.description }),
|
||||
value: { outcome: "selected" as const, index },
|
||||
})),
|
||||
{ label: "Write a custom answer", value: { outcome: "custom" as const } },
|
||||
],
|
||||
);
|
||||
|
||||
let response: QuestionResponse;
|
||||
if (choice === undefined) {
|
||||
response = { outcome: "cancelled" };
|
||||
} else if (choice.outcome === "selected") {
|
||||
response = choice;
|
||||
} else {
|
||||
const answer = await modal.input(request.question);
|
||||
response = answer === undefined ? { outcome: "cancelled" } : { outcome: "custom", answer };
|
||||
}
|
||||
|
||||
await dialog.service.submitAnswer(response, context);
|
||||
} finally {
|
||||
modal.close();
|
||||
let response: QuestionResponse;
|
||||
if (choice === undefined) {
|
||||
response = { outcome: "cancelled" };
|
||||
} else if (choice.outcome === "selected") {
|
||||
response = choice;
|
||||
} else {
|
||||
const answer = await modal.input(request.question);
|
||||
response = answer === undefined ? { outcome: "cancelled" } : { outcome: "custom", answer };
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
await dialog.submitAnswer(response, context);
|
||||
} finally {
|
||||
modal.close();
|
||||
}
|
||||
});
|
||||
|
||||
tui.toolRenderers.add<QuestionDetails>("question", questionRenderer);
|
||||
},
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
import { FacetKernel } from "./facets/host.ts";
|
||||
import { disposeLoadedFacets } from "./facets/loader.ts";
|
||||
import { RemoteServiceBindingImpl } from "./services/consumer.ts";
|
||||
import { lookupServiceInstanceKey } from "./services/instances.ts";
|
||||
import type { RemoteServiceProvider } from "./services/provider.ts";
|
||||
import { MutableReplicatedStateImpl, serviceDeliveryContext } from "./services/state.ts";
|
||||
import { MutableReplicatedStateImpl } from "./services/state.ts";
|
||||
import type {
|
||||
Facet,
|
||||
FacetHost,
|
||||
@@ -13,7 +11,6 @@ import type {
|
||||
MutableReplicatedState,
|
||||
RemoteServiceBinding,
|
||||
RemoteServiceBindingOptions,
|
||||
RemoteServiceConnection,
|
||||
RemoteServiceContract,
|
||||
Service,
|
||||
} from "./types.ts";
|
||||
@@ -89,32 +86,6 @@ export function createRemoteServiceBinding(options: RemoteServiceBindingOptions)
|
||||
return new RemoteServiceBindingImpl(options);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the application key for a keyed service proxy, or undefined for a singleton.
|
||||
*
|
||||
* @publicApiReview Consider passing the key directly to keyed observers instead of exposing proxy metadata.
|
||||
*/
|
||||
export function getServiceInstanceKey(service: object): string | undefined {
|
||||
return lookupServiceInstanceKey(service);
|
||||
}
|
||||
|
||||
/** @publicApiReview Consider keeping this loopback transport adapter in testing support. */
|
||||
export function createLoopbackServiceConnection(provider: RemoteServiceProvider): RemoteServiceConnection {
|
||||
return {
|
||||
invoke: (call, context) => provider.invoke(call, context),
|
||||
subscribe: async (serviceId, mode, listener) => {
|
||||
const subscription = provider.subscribe(serviceId, mode, (update) =>
|
||||
listener(update, serviceDeliveryContext()),
|
||||
);
|
||||
return {
|
||||
snapshot: subscription.snapshot,
|
||||
activate: () => subscription.activate(),
|
||||
close: () => subscription.close(),
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function replicatedState<T>(initial: T): MutableReplicatedState<T> {
|
||||
return new MutableReplicatedStateImpl(initial);
|
||||
}
|
||||
|
||||
@@ -119,11 +119,6 @@ class FacetLifecycle {
|
||||
}
|
||||
}
|
||||
|
||||
interface StagedObservation {
|
||||
start(): () => void;
|
||||
stop(): void;
|
||||
}
|
||||
|
||||
interface KeyedServiceSource {
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): () => void;
|
||||
}
|
||||
@@ -226,23 +221,11 @@ class FacetServiceHandles {
|
||||
return handle.proxy as T;
|
||||
}
|
||||
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): StagedObservation {
|
||||
let close: (() => void) | undefined;
|
||||
let closed = false;
|
||||
const stop = (): void => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
close?.();
|
||||
};
|
||||
return {
|
||||
start: () => {
|
||||
if (closed) return () => {};
|
||||
const source = this.#keyedSources.get(service.id);
|
||||
if (source === undefined) throw new Error(`Service ${service.id} is disconnected`);
|
||||
close = source.observe(service, handler);
|
||||
return stop;
|
||||
},
|
||||
stop,
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): () => Disposal {
|
||||
return () => {
|
||||
const source = this.#keyedSources.get(service.id);
|
||||
if (source === undefined) throw new Error(`Service ${service.id} is disconnected`);
|
||||
return source.observe(service, handler);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -390,7 +373,7 @@ export class FacetKernel {
|
||||
|
||||
this.#phase = "connecting";
|
||||
await Promise.all(
|
||||
[...this.#connectionBindings.values()].map((services) => services.activate(BACKGROUND_CONTEXT)),
|
||||
[...this.#connectionBindings.values()].map((services) => services.ready(BACKGROUND_CONTEXT)),
|
||||
);
|
||||
|
||||
this.#phase = "activating";
|
||||
@@ -575,15 +558,10 @@ export class FacetKernel {
|
||||
recordServiceReference(runtime.requires, service, "singleton");
|
||||
return this.#handles.use(service);
|
||||
},
|
||||
observe: <T>(
|
||||
service: Service<T>,
|
||||
handler: (service: T, context: Context) => void | Promise<void>,
|
||||
): (() => void) => {
|
||||
observe: <T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): void => {
|
||||
lifecycle.assertSettingUp("observe services");
|
||||
recordServiceReference(runtime.requires, service, "keyed");
|
||||
const observation = this.#handles.observe(service, handler);
|
||||
lifecycle.observe(() => observation.start());
|
||||
return observation.stop;
|
||||
lifecycle.observe(this.#handles.observe(service, handler));
|
||||
},
|
||||
replicatedState: <T>(initial: T) => {
|
||||
lifecycle.assertRunning("create replicated state");
|
||||
|
||||
@@ -4,12 +4,10 @@
|
||||
export {
|
||||
combineFacetLoaders,
|
||||
createFacetHost,
|
||||
createLoopbackServiceConnection,
|
||||
createRemoteServiceBinding,
|
||||
createStaticFacetLoader,
|
||||
defineFacet,
|
||||
defineService,
|
||||
getServiceInstanceKey,
|
||||
replicatedState,
|
||||
} from "./api.ts";
|
||||
export { RemoteServiceError } from "./services/errors.ts";
|
||||
|
||||
@@ -410,7 +410,7 @@ export class RemoteServiceBindingImpl implements RemoteServiceBinding {
|
||||
readonly #allowlist = new Set<string>();
|
||||
readonly #reportError: ErrorReporter;
|
||||
readonly #modes = new Map<string, "singleton" | "keyed">();
|
||||
#assertAccess: () => void;
|
||||
readonly #assertAccess: () => void;
|
||||
readonly #singletons = new Map<string, SingletonBinding>();
|
||||
readonly #keyed = new Map<string, KeyedBinding<unknown>>();
|
||||
#bound: boolean;
|
||||
@@ -481,14 +481,6 @@ export class RemoteServiceBindingImpl implements RemoteServiceBinding {
|
||||
return binding.observe(handler);
|
||||
}
|
||||
|
||||
setAccessGuard(assertAccess: () => void): void {
|
||||
this.#assertAccess = assertAccess;
|
||||
}
|
||||
|
||||
async activate(context: Context): Promise<void> {
|
||||
await this.ready(context);
|
||||
}
|
||||
|
||||
async ready(context: Context): Promise<void> {
|
||||
if (this.#disposed) throw new Error("Remote service binding is disposed");
|
||||
while (true) {
|
||||
|
||||
@@ -1,12 +1,6 @@
|
||||
import { BACKGROUND_CONTEXT, withCancel } from "../context/index.ts";
|
||||
import type { Context } from "../types.ts";
|
||||
|
||||
const instanceKeys = new WeakMap<object, string>();
|
||||
|
||||
export function lookupServiceInstanceKey(service: object): string | undefined {
|
||||
return instanceKeys.get(service);
|
||||
}
|
||||
|
||||
export interface InstanceDirectoryEntry {
|
||||
readonly key: string;
|
||||
readonly generation: number;
|
||||
@@ -47,7 +41,6 @@ export class InstanceDirectory<TEntry extends InstanceDirectoryEntry> {
|
||||
throw new Error(`Keyed service already has a live instance with key ${entry.key}`);
|
||||
}
|
||||
this.#entries.set(entry.key, entry);
|
||||
registerServiceInstance(entry.service, entry.key);
|
||||
if (this.#ready) this.#startAll(entry);
|
||||
}
|
||||
|
||||
@@ -61,7 +54,6 @@ export class InstanceDirectory<TEntry extends InstanceDirectoryEntry> {
|
||||
this.#remove(previous);
|
||||
}
|
||||
this.#entries.set(entry.key, entry);
|
||||
registerServiceInstance(entry.service, entry.key);
|
||||
if (this.#ready) this.#startAll(entry);
|
||||
}
|
||||
|
||||
@@ -150,10 +142,6 @@ export class InstanceDirectory<TEntry extends InstanceDirectoryEntry> {
|
||||
}
|
||||
}
|
||||
|
||||
function registerServiceInstance(service: object, key: string): void {
|
||||
instanceKeys.set(service, key);
|
||||
}
|
||||
|
||||
function toError(error: unknown): Error {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
|
||||
@@ -93,12 +93,8 @@ export interface ServiceSpawner<T> {
|
||||
export interface RemoteServices {
|
||||
use<T>(service: Service<T>): T;
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): () => void;
|
||||
/** Activate this service set and wait for every acquired service to become ready. */
|
||||
activate(context: Context): Promise<void>;
|
||||
/** Wait until every currently acquired service has installed its initial snapshot. */
|
||||
ready(context: Context): Promise<void>;
|
||||
/** Install a host lifecycle guard for proxy member access. */
|
||||
setAccessGuard(assertAccess: () => void): void;
|
||||
dispose(context: Context): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -207,7 +203,7 @@ export interface FacetEnvironment {
|
||||
/** Declare a hard dependency on one singleton service and return its stable handle. */
|
||||
use<T>(service: Service<T>): T;
|
||||
/** Declare a hard dependency on a keyed service and observe each live instance. */
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): () => void;
|
||||
observe<T>(service: Service<T>, handler: (service: T, context: Context) => void | Promise<void>): void;
|
||||
/** Declare and install this facet's singleton implementation of a service. */
|
||||
provide<T>(service: Service<T>, implementation: NoInfer<T>): void;
|
||||
/** Declare ownership of a multi-instance service and return its deferred spawning capability. */
|
||||
|
||||
@@ -4,13 +4,13 @@ import {
|
||||
type Context,
|
||||
combineFacetLoaders,
|
||||
createFacetHost,
|
||||
createLoopbackServiceConnection,
|
||||
createRemoteServiceBinding,
|
||||
createStaticFacetLoader,
|
||||
defineFacet,
|
||||
defineService,
|
||||
type FacetLoader,
|
||||
} from "../src/index.ts";
|
||||
import { createLoopbackServiceConnection } from "./helpers.ts";
|
||||
|
||||
interface GenerationValue {
|
||||
read(context: Context): Promise<string>;
|
||||
|
||||
@@ -3,7 +3,6 @@ import { BACKGROUND_CONTEXT } from "../src/context/index.ts";
|
||||
import {
|
||||
type Context,
|
||||
createFacetHost,
|
||||
createLoopbackServiceConnection,
|
||||
createRemoteServiceBinding,
|
||||
defineFacet,
|
||||
defineService,
|
||||
@@ -12,6 +11,7 @@ import {
|
||||
RemoteServiceProvider,
|
||||
type ReplicatedState,
|
||||
} from "../src/index.ts";
|
||||
import { createLoopbackServiceConnection } from "./helpers.ts";
|
||||
|
||||
interface Source {
|
||||
read(context: Context): Promise<string>;
|
||||
@@ -283,8 +283,9 @@ describe("facet host", () => {
|
||||
const connections = [
|
||||
{ namespace: leftNamespace, provider: leftProvider },
|
||||
{ namespace: rightNamespace, provider: rightProvider },
|
||||
].map(({ namespace, provider }) =>
|
||||
Object.assign(namespace, {
|
||||
].map(({ namespace, provider }) => {
|
||||
const ready = namespace.ready.bind(namespace);
|
||||
return Object.assign(namespace, {
|
||||
acceptsUnavailableServices: false,
|
||||
async catalogue() {
|
||||
return provider.catalogue;
|
||||
@@ -292,12 +293,12 @@ describe("facet host", () => {
|
||||
open() {
|
||||
return namespace;
|
||||
},
|
||||
async activate(context: Context) {
|
||||
async ready(context: Context) {
|
||||
await namespace.rebind(true, context);
|
||||
await namespace.ready(context);
|
||||
await ready(context);
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
const facet = defineFacet({
|
||||
id: "combined",
|
||||
setup(env) {
|
||||
@@ -371,11 +372,12 @@ describe("facet host", () => {
|
||||
assertAccess: options.assertAccess,
|
||||
onError: options.onError,
|
||||
});
|
||||
const ready = namespace.ready.bind(namespace);
|
||||
const dispose = namespace.dispose.bind(namespace);
|
||||
return Object.assign(namespace, {
|
||||
async activate(context: Context) {
|
||||
async ready(context: Context) {
|
||||
await namespace.rebind(true, context);
|
||||
await namespace.ready(context);
|
||||
await ready(context);
|
||||
},
|
||||
async dispose(context: Context) {
|
||||
disposed += 1;
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
import type { RemoteServiceConnection } from "../src/index.ts";
|
||||
import type { RemoteServiceProvider } from "../src/services/provider.ts";
|
||||
import { serviceDeliveryContext } from "../src/services/state.ts";
|
||||
|
||||
export function createLoopbackServiceConnection(provider: RemoteServiceProvider): RemoteServiceConnection {
|
||||
return {
|
||||
invoke: (call, context) => provider.invoke(call, context),
|
||||
subscribe: async (serviceId, mode, listener) => {
|
||||
const subscription = provider.subscribe(serviceId, mode, (update) =>
|
||||
listener(update, serviceDeliveryContext()),
|
||||
);
|
||||
return {
|
||||
snapshot: subscription.snapshot,
|
||||
activate: () => subscription.activate(),
|
||||
close: () => subscription.close(),
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -2,16 +2,15 @@ import { describe, expect, test, vi } from "vitest";
|
||||
import { BACKGROUND_CONTEXT } from "../src/context/index.ts";
|
||||
import {
|
||||
type Context,
|
||||
createLoopbackServiceConnection,
|
||||
createRemoteServiceBinding,
|
||||
defineService,
|
||||
getServiceInstanceKey,
|
||||
type JsonValue,
|
||||
type RemoteServiceConnection,
|
||||
RemoteServiceProvider,
|
||||
type ReplicatedState,
|
||||
replicatedState,
|
||||
} from "../src/index.ts";
|
||||
import { createLoopbackServiceConnection } from "./helpers.ts";
|
||||
|
||||
type ModelRef = { provider: string; modelId: string };
|
||||
type ModelsState = {
|
||||
@@ -344,14 +343,12 @@ describe("remote services", () => {
|
||||
onError: (error) => errors.push(error),
|
||||
});
|
||||
const observed: {
|
||||
key: string;
|
||||
question: Question | undefined;
|
||||
service: QuestionDialogs;
|
||||
context: Context;
|
||||
}[] = [];
|
||||
const stop = namespace.observe(QuestionDialogs, (service, context) => {
|
||||
observed.push({
|
||||
key: getServiceInstanceKey(service)!,
|
||||
question: service.request.value,
|
||||
service,
|
||||
context,
|
||||
@@ -366,7 +363,7 @@ describe("remote services", () => {
|
||||
submit: firstSubmit,
|
||||
});
|
||||
await vi.waitFor(() => expect(observed).toHaveLength(1));
|
||||
expect(observed[0]).toMatchObject({ key: "invocation-1", question: { question: "First?" } });
|
||||
expect(observed[0]).toMatchObject({ question: { question: "First?" } });
|
||||
|
||||
const firstService = observed[0]!.service;
|
||||
firstRequest.set({ question: "Updated?" }, BACKGROUND_CONTEXT);
|
||||
@@ -388,7 +385,7 @@ describe("remote services", () => {
|
||||
},
|
||||
});
|
||||
await vi.waitFor(() => expect(observed).toHaveLength(2));
|
||||
expect(observed[1]).toMatchObject({ key: "invocation-1", question: { question: "Again?" } });
|
||||
expect(observed[1]).toMatchObject({ question: { question: "Again?" } });
|
||||
expect(observed[1]!.service).not.toBe(firstService);
|
||||
expect(errors).toEqual([]);
|
||||
|
||||
|
||||
@@ -173,7 +173,7 @@ export async function activateBuiltinClientServices(
|
||||
};
|
||||
const models = sessionServices.use(Models);
|
||||
const agent = sessionServices.use(AgentController);
|
||||
await Promise.all([serverServices.activate(BACKGROUND_CONTEXT), sessionServices.activate(BACKGROUND_CONTEXT)]);
|
||||
await Promise.all([serverServices.ready(BACKGROUND_CONTEXT), sessionServices.ready(BACKGROUND_CONTEXT)]);
|
||||
return { ...server, directory, management, models, agent };
|
||||
}
|
||||
|
||||
|
||||
@@ -58,6 +58,7 @@ class RoutedServiceBinding implements RemoteServices {
|
||||
readonly #onActivate: (context: Context) => Promise<void>;
|
||||
readonly #remove: () => void;
|
||||
#activated = false;
|
||||
#activationComplete = false;
|
||||
|
||||
constructor(options: {
|
||||
readonly services: readonly { readonly id: string }[];
|
||||
@@ -88,20 +89,19 @@ class RoutedServiceBinding implements RemoteServices {
|
||||
return this.#services.observe(service, handler);
|
||||
}
|
||||
|
||||
setAccessGuard(assertAccess: () => void): void {
|
||||
this.#services.setAccessGuard(assertAccess);
|
||||
}
|
||||
|
||||
async activate(context: Context): Promise<void> {
|
||||
async ready(context: Context): Promise<void> {
|
||||
if (!this.#activated) {
|
||||
this.#activated = true;
|
||||
await this.#services.rebind(this.#getBound(), context);
|
||||
}
|
||||
await this.#services.ready(context);
|
||||
await this.#onActivate(context);
|
||||
await this.serviceReady(context);
|
||||
if (!this.#activationComplete) {
|
||||
await this.#onActivate(context);
|
||||
this.#activationComplete = true;
|
||||
}
|
||||
}
|
||||
|
||||
ready(context: Context): Promise<void> {
|
||||
serviceReady(context: Context): Promise<void> {
|
||||
return this.#services.ready(context);
|
||||
}
|
||||
|
||||
@@ -323,7 +323,7 @@ class SessionServiceConnectionImpl implements SessionServiceConnection {
|
||||
|
||||
async #whenAttached(attachment: SessionTarget, revision: number, context: Context): Promise<void> {
|
||||
try {
|
||||
await Promise.all([...this.#bindings].map((binding) => binding.ready(context)));
|
||||
await Promise.all([...this.#bindings].map((binding) => binding.serviceReady(context)));
|
||||
} catch (error) {
|
||||
if (this.#attachmentRevision === revision && sameAttachment(this.#client.attachment, attachment)) {
|
||||
this.#attachmentState.set({ status: "degraded", sessionId: attachment.sessionId }, context);
|
||||
@@ -340,7 +340,7 @@ class SessionServiceConnectionImpl implements SessionServiceConnection {
|
||||
}
|
||||
|
||||
async #whenDetached(revision: number, context: Context): Promise<void> {
|
||||
await Promise.all([...this.#bindings].map((binding) => binding.ready(context)));
|
||||
await Promise.all([...this.#bindings].map((binding) => binding.serviceReady(context)));
|
||||
if (this.#attachmentRevision !== revision || this.#client.attachment !== undefined) {
|
||||
throw new Error("The Session attachment changed while detaching");
|
||||
}
|
||||
@@ -361,7 +361,7 @@ export function createServerServiceBinding(
|
||||
assertAccess() {},
|
||||
onError: options.onError ?? (() => {}),
|
||||
});
|
||||
const activation = services.activate(BACKGROUND_CONTEXT);
|
||||
const activation = services.ready(BACKGROUND_CONTEXT);
|
||||
void activation.catch(options.onError ?? (() => {}));
|
||||
const ready = async (context: Context): Promise<void> => {
|
||||
await activation;
|
||||
@@ -374,9 +374,7 @@ export function createServerServiceBinding(
|
||||
open: (openOptions) => connection.open(openOptions),
|
||||
use: (service) => services.use(service),
|
||||
observe: (service, handler) => services.observe(service, handler),
|
||||
activate: ready,
|
||||
ready,
|
||||
setAccessGuard: (assertAccess) => services.setAccessGuard(assertAccess),
|
||||
async dispose(context) {
|
||||
const results = await Promise.allSettled([services.dispose(context), connection.dispose(context)]);
|
||||
throwFailures(results, "Failed to dispose server service binding");
|
||||
@@ -395,7 +393,7 @@ export function createSessionServiceBinding(
|
||||
assertAccess() {},
|
||||
onError: options.onError ?? (() => {}),
|
||||
});
|
||||
const activation = services.activate(BACKGROUND_CONTEXT);
|
||||
const activation = services.ready(BACKGROUND_CONTEXT);
|
||||
void activation.catch(options.onError ?? (() => {}));
|
||||
const ready = async (context: Context): Promise<void> => {
|
||||
await activation;
|
||||
@@ -412,9 +410,7 @@ export function createSessionServiceBinding(
|
||||
open: (openOptions) => connection.open(openOptions),
|
||||
use: (service) => services.use(service),
|
||||
observe: (service, handler) => services.observe(service, handler),
|
||||
activate: ready,
|
||||
ready,
|
||||
setAccessGuard: (assertAccess) => services.setAccessGuard(assertAccess),
|
||||
whenAttached: (sessionId, context) => connection.whenAttached(sessionId, context),
|
||||
whenDetached: (context) => connection.whenDetached(context),
|
||||
async dispose(context) {
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import {
|
||||
createLoopbackServiceConnection,
|
||||
createRemoteServiceBinding,
|
||||
defineFacet,
|
||||
type FacetLoader,
|
||||
type RemoteServiceConnection,
|
||||
RemoteServiceProvider,
|
||||
replicatedState,
|
||||
} from "@earendil-works/chord";
|
||||
@@ -34,6 +34,20 @@ function session(sessionId: string, createdAt: number): SessionSummary {
|
||||
return { serverId, sessionId, createdAt };
|
||||
}
|
||||
|
||||
function createLoopbackServiceConnection(provider: RemoteServiceProvider): RemoteServiceConnection {
|
||||
return {
|
||||
invoke: (call, context) => provider.invoke(call, context),
|
||||
subscribe: async (serviceId, mode, listener) => {
|
||||
const subscription = provider.subscribe(serviceId, mode, (update) => listener(update, BACKGROUND_CONTEXT));
|
||||
return {
|
||||
snapshot: subscription.snapshot,
|
||||
activate: () => subscription.activate(),
|
||||
close: () => subscription.close(),
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function laneSnapshot(): LaneSnapshot {
|
||||
return {
|
||||
lane: "main",
|
||||
@@ -196,6 +210,7 @@ describe("experimental client TUI", () => {
|
||||
connection: createLoopbackServiceConnection(serverProvider),
|
||||
bound: false,
|
||||
});
|
||||
const serverNamespaceReady = serverNamespace.ready.bind(serverNamespace);
|
||||
const serverServices: ServerServiceConnection = Object.assign(serverNamespace, {
|
||||
acceptsUnavailableServices: false,
|
||||
connection: connectionState,
|
||||
@@ -205,9 +220,9 @@ describe("experimental client TUI", () => {
|
||||
open() {
|
||||
return serverNamespace;
|
||||
},
|
||||
async activate() {
|
||||
async ready() {
|
||||
await serverNamespace.rebind(true, BACKGROUND_CONTEXT);
|
||||
await serverNamespace.ready(BACKGROUND_CONTEXT);
|
||||
await serverNamespaceReady(BACKGROUND_CONTEXT);
|
||||
},
|
||||
});
|
||||
const sessionNamespace = createRemoteServiceBinding({
|
||||
@@ -224,9 +239,6 @@ describe("experimental client TUI", () => {
|
||||
open() {
|
||||
return sessionNamespace;
|
||||
},
|
||||
async activate() {
|
||||
await sessionNamespace.ready(BACKGROUND_CONTEXT);
|
||||
},
|
||||
async whenAttached(sessionId: string) {
|
||||
await sessionNamespace.rebind(true, BACKGROUND_CONTEXT);
|
||||
await sessionNamespace.ready(BACKGROUND_CONTEXT);
|
||||
|
||||
Reference in New Issue
Block a user