Function createMessageReader

packages/cli/src/mcp/bridge-e2e.test.ts:38–100

(stdout: PassThrough)

}

function createMessageReader(stdout: PassThrough): {
  waitFor: (
    predicate: (message: Record<string, unknown>) => boolean,
  ) => Promise<Record<string, unknown>>;
} {
  let buffer = '';
  const messages: Record<string, unknown>[] = [];
  const waiters: Array<{
    predicate: (message: Record<string, unknown>) => boolean;
    resolve: (message: Record<string, unknown>) => void;
    reject: (err: Error) => void;
    timer: ReturnType<typeof setTimeout>;
  }> = [];

  function drainWaiters(): void {
    for (let waiterIndex = 0; waiterIndex < waiters.length; waiterIndex++) {
      const waiter = waiters[waiterIndex];
      if (!waiter) continue;
      const messageIndex = messages.findIndex(waiter.predicate);
      if (messageIndex === -1) continue;
      const [message] = messages.splice(messageIndex, 1);
      clearTimeout(waiter.timer);
      waiters.splice(waiterIndex, 1);
      waiter.resolve(message);
      waiterIndex--;
    }
  }

  stdout.on('data', (chunk: Buffer) => {
    buffer += chunk.toString('utf8');
    while (true) {
      const newline = buffer.indexOf('\n');
      if (newline === -1) break;
      const line = buffer.slice(0, newline).replace(/\r$/, '');
      buffer = buffer.slice(newline + 1);
      if (line.length === 0) continue;
      const parsed = JSON.parse(line) as unknown;
      if (!isRecord(parsed)) {
        throw new Error(`stdio response was not a JSON object: ${line}`);
      }
      messages.push(parsed);
    }
    drainWaiters();
  });

  return {
    waitFor(predicate) {
      const existingIndex = messages.findIndex(predicate);
      if (existingIndex !== -1) {
        const [message] = messages.splice(existingIndex, 1);
        return Promise.resolve(message);
      }
      return new Promise((resolve, reject) => {
        const timer = setTimeout(() => {
          const index = waiters.findIndex((waiter) => waiter.timer === timer);
          if (index !== -1) waiters.splice(index, 1);
          reject(new Error('timed out waiting for stdio JSON-RPC response'));
        }, 5_000);