Class RuntimeRemoteBridgeWorker

runtime/api-server/src/bridge-worker.ts:389–687

}

export class RuntimeRemoteBridgeWorker implements BridgeWorkerLike {
  readonly #logger: LoggerLike | undefined;
  readonly #executeJob: (job: ProactiveBridgeJob) => Promise<ProactiveBridgeJobResult>;
  readonly #captureRuntimeException: typeof captureRuntimeException;
  readonly #fetch: typeof fetch;
  readonly #baseUrl: string;
  readonly #pollIntervalMs: number;
  readonly #maxItems: number;
  readonly #headers: Record<string, string>;
  #stopped = false;
  #task: Promise<void> | null = null;
  #wakeResolver: (() => void) | null = null;

  constructor(options: RuntimeRemoteBridgeWorkerOptions = {}) {
    this.#logger = options.logger;
    this.#executeJob =
      options.executeJob ??
      (options.store
        ? (job) => executeBridgeJobNatively({ job, store: options.store as RuntimeStateStore })
        : (() => {
            throw new Error("bridge worker requires executeJob or store");
          }));
    this.#captureRuntimeException =
      options.captureRuntimeException ?? captureRuntimeException;
    this.#fetch = options.fetchImpl ?? fetch;
    this.#baseUrl = options.baseUrl ?? proactiveBridgeBaseUrl();
    this.#pollIntervalMs = options.pollIntervalMs ?? bridgePollIntervalMs();
    this.#maxItems = options.maxItems ?? bridgeMaxItems();
    this.#headers = proactiveBridgeHeaders();
  }

  async start(): Promise<void> {
    if (this.#task) {
      return;
    }
    this.#stopped = false;
    this.#task = this.#runLoop();
  }

  async close(): Promise<void> {
    this.#stopped = true;
    const resolve = this.#wakeResolver;
    this.#wakeResolver = null;
    resolve?.();
    const task = this.#task;
    this.#task = null;
    await task;
  }

  async pollOnce(): Promise<number> {
    let jobs: ProactiveBridgeJob[];
    try {
      jobs = await this.#receiveJobs();
    } catch (error) {
      this.#capturePollFailure(error);
      throw error;
    }
    for (const job of jobs) {