Method handleStreamingMessage

src/a2a-mock.ts:254–373

(
    parsed: Record<string, unknown>,
    req: http.IncomingMessage,
    res: http.ServerResponse,
  )

  // ---- Private: streaming handler ----

  private async handleStreamingMessage(
    parsed: Record<string, unknown>,
    req: http.IncomingMessage,
    res: http.ServerResponse,
  ): Promise<void> {
    const params = parsed.params as Record<string, unknown> | undefined;
    const id = parsed.id as string | number;
    const text = extractText(params);
    const entry = findStreamingMatch(text, this.agents);

    if (!entry) {
      res.writeHead(200, {
        "Content-Type": "application/json",
        "A2A-Version": "1.0",
      });
      res.end(
        JSON.stringify({
          jsonrpc: "2.0",
          id,
          error: { code: -32000, message: "No matching pattern for message" },
        }),
      );
      return;
    }

    // Create task for the streaming response
    const taskId = generateId("task");
    const contextId = generateId("ctx");
    const userParts: A2APart[] = params?.message
      ? (((params.message as Record<string, unknown>).parts as A2APart[]) ?? [{ text }])
      : [{ text }];

    const task: A2ATask = {
      id: taskId,
      contextId,
      status: { state: "TASK_STATE_WORKING", timestamp: new Date().toISOString() },
      artifacts: [],
      history: [
        {
          messageId: generateId("msg"),
          role: "ROLE_USER",
          parts: userParts,
        },
      ],
    };
    this.tasks.set(taskId, task);

    // Write SSE response
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
      "A2A-Version": "1.0",
    });

    const delayMs = entry.delayMs ?? 0;

    for (const event of entry.events) {