runtimes/core: stream service to service

[fredr]

Adds a service to service stream client in core and adds it to runtimes/js

[encore-cla]

All committers have signed the CLA.

[fredr]

This also makes it possible to test streams in two ways, either directly by testing the handler and mocking the stream, or by using the service to service client.

Some example on how tests could be written from my testing, to gice some context

import { describe, expect, test, vi } from "vitest";
import { CalcMessage, calc, InMessage, OutMessage, bidi } from "./api";
import { streams } from "~encore/clients";

describe("calc", () => {
  test("with the imported handler", async () => {
    const streamMock = {
      recvMessages: [
        { operation: { type: "add", value: 1 } },
        { operation: { type: "add", value: 2 } },
        { operation: { type: "add", value: 3 } },
        { operation: { type: "end" } },
      ],
      async recv(): Promise<any> {
        return this.recvMessages.shift();
      },
      async close() {},
      async *[Symbol.asyncIterator]() {
        while (true) {
          try {
            yield await this.recv();
          } catch (e) {
            break;
          }
        }
      },
    };

    const { value } = await calc(streamMock);
    expect(value).toBe(6);
  });

  test("imported handler with vitest mock", async () => {
    const mockedStream = {
      recv: vi.fn<[], Promise<CalcMessage>>(),
      close: vi.fn<[], Promise<void>>(),
      async *[Symbol.asyncIterator]() {
        while (true) {
          try {
            yield await this.recv();
          } catch (e) {
            break;
          }
        }
      },
    };

    mockedStream.recv.mockResolvedValueOnce({
      operation: { type: "add", value: 1 },
    });
    mockedStream.recv.mockResolvedValueOnce({
      operation: { type: "add", value: 5 },
    });
    mockedStream.recv.mockResolvedValueOnce({
      operation: { type: "add", value: 4 },
    });
    mockedStream.recv.mockResolvedValueOnce({ operation: { type: "end" } });

    const { value } = await calc(mockedStream);
    expect(value).toBe(10);
  });

  test("imported inout handler with vitest mock", async () => {
    const mockedStream = {
      recv: vi.fn<[], Promise<InMessage>>(),
      send: vi.fn<[OutMessage], Promise<void>>(),
      close: vi.fn<[], Promise<void>>(),
      async *[Symbol.asyncIterator]() {
        while (true) {
          try {
            yield await this.recv();
          } catch (e) {
            break;
          }
        }
      },
    };

    mockedStream.recv.mockResolvedValueOnce({ value: "one" });
    mockedStream.recv.mockResolvedValueOnce({ value: "two" });
    mockedStream.recv.mockResolvedValueOnce({ value: "three" });
    mockedStream.recv.mockResolvedValueOnce({ value: "four" });
    mockedStream.recv.mockResolvedValueOnce({ value: "five" });
    mockedStream.recv.mockResolvedValueOnce({ value: "six" });
    mockedStream.recv.mockResolvedValueOnce({ value: "seven" });
    mockedStream.recv.mockResolvedValueOnce({ value: "eight" });
    mockedStream.recv.mockResolvedValueOnce({ value: "nine" });
    mockedStream.recv.mockResolvedValueOnce({ value: "ten" });
    mockedStream.recv.mockResolvedValueOnce({ value: "eleven" });

    await bidi(mockedStream);

    expect(mockedStream.send).toHaveBeenNthCalledWith(1, { value: 10 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(2, { value: 9 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(3, { value: 8 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(4, { value: 7 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(5, { value: 6 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(6, { value: 5 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(7, { value: 4 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(8, { value: 3 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(9, { value: 2 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(10, { value: 1 });
    expect(mockedStream.send).toHaveBeenNthCalledWith(11, { value: 0 });
  });

  test("with service to service in stream client", async () => {
    const stream = await streams.calc();
    await stream.send({ operation: { type: "add", value: 5 } });
    await stream.send({ operation: { type: "add", value: 5 } });
    await stream.send({ operation: { type: "add", value: 5 } });
    await stream.send({ operation: { type: "end" } });
    const resp = await stream.response();

    expect(resp.value).toBe(15);
  });

  test("with service to service inout stream client", async () => {
    const stream = await streams.bidi();
    for (var i = 0; i <= 10; i++) {
      await stream.send({ value: "hello there" });
      const { value } = await stream.recv();
      expect(value).toBe(10 - i);
    }
  });

  test("with service to service out stream client", async () => {
    const stream = await streams.logs({ amount: 5 });
    let numLogs = 0;

    for await (const _log of stream) {
      numLogs += 1;
    }

    expect(numLogs).toBe(5);
  });
});