Method callStream

node/src/transport.ts:314–348

Open a streaming call (machines.ask). Returns an AskStream.

(req: Envelope)

  /** Open a streaming call (machines.ask). Returns an {@link AskStream}. */
  callStream(req: Envelope): AskStream {
    const queue = new FrameQueue();
    let started = false;
    let id = 0;
    const transport = this;

    const source: StreamSource = {
      get id() {
        return id;
      },
      async *envelopes() {
        if (!started) {
          started = true;
          await transport.ensureStarted();
          id = transport.#nextId++;
          req.id = id;
          req.kind = Envelope_Kind.REQUEST;
          transport.#pending.set(id, {
            kind: "stream",
            push: (e) => queue.push(e),
            end: () => queue.end(),
            fail: (e) => queue.fail(e),
          });
          transport.#write(req);
        }
        yield* queue.iterate();
      },
      async answer(env: Envelope) {
        await transport.ensureStarted();
        transport.#write(env);
      },
    };

    return new AskStream(source);
  }
}