feuerwerk/tests/viewport/streaming.test.ts

/**
 * Streaming a sound (#349) on the page's side: the copy fetched a piece at a time, its first seconds
 * decoded ahead, and a `stream-player` kept fed along its plan – every sample where it belongs, at a
 * jump as well as round a loop, and no further ahead than it needs.
 */
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { StreamPlan } from '../../src/core/music/streams';
import { StreamFeed, StreamFile } from '../../src/viewport/streaming';
import { decoders, fakeCopy, PRE_SKIP, serveCopies, stubDecoder } from './fakeStream';

const URL = '/sounds/copy.ogg';
const settle = () => new Promise((resolve) => setTimeout(resolve, 0));

/** A worklet node as the feed sees it: what was posted to it, and a way to report what it read. */
function fakeNode() {
  const posted: unknown[] = [];
  const node = {
    port: {
      onmessage: null as ((event: { data: unknown }) => void) | null,
      postMessage: (data: unknown) => posted.push(data),
      close: () => {},
    },
    disconnect: () => {},
  };
  const samples = () =>
    (posted.filter((data) => Array.isArray(data) && data.length > 0) as Float32Array[][]).flatMap((chunk) =>
      Array.from(chunk[0]!),
    );
  const read = (used: number) => node.port.onmessage?.({ data: { used, skipped: 0 } });
  return { node: node as unknown as AudioWorkletNode, posted, samples, read };
}

/** A feed as the library makes it: its opening handed over first, then attached to the player. */
function feedOf(file: StreamFile, plan: StreamPlan, fake: ReturnType<typeof fakeNode>, ended = () => {}) {
  const feed = new StreamFeed(file, plan, ended);
  fake.posted.push(feed.opening());
  feed.attach(fake.node);
  return feed;
}

/** Waits until the feed has nothing left to do for now. */
async function drain(): Promise<void> {
  for (let i = 0; i < 400; i++) await settle();
}

beforeEach(() => stubDecoder());
afterEach(() => vi.unstubAllGlobals());

describe('a copy streamed', () => {
  it('fetches pieces until its first two seconds are there and decodes them, its pre-skip cut off', async () => {
    const copy = fakeCopy(10);
    const server = serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    expect(await file.ready()).toBe(true);
    expect(file.head![0]!.length).toBe(96000);
    expect([file.head![0]![0], file.head![0]![95999], file.head![1]![0]]).toEqual([
      PRE_SKIP,
      PRE_SKIP + 95999,
      PRE_SKIP + 1e6,
    ]);
    expect(server.asked[0]).toBe('bytes=0-262143');
    expect(file.bytes).toBeLessThan(copy.bytes.length);
  });

  it('takes the whole file from a server that leaves `Range` out', async () => {
    const copy = fakeCopy(5);
    const server = serveCopies({ [URL]: copy.bytes }, false);
    const file = new StreamFile(URL, copy.frames);
    expect(await file.ready()).toBe(true);
    expect(server.asked).toHaveLength(1);
    expect(file.bytes).toBe(copy.bytes.length);
  });

  it('sends its head at once, then decodes on from there to the sample, no more than four seconds ahead', async () => {
    const copy = fakeCopy(10);
    serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    await file.ready();
    const player = fakeNode();
    const { samples, read } = player;
    const feed = new StreamFeed(file, { from: 0, until: copy.frames, loop: null }, () => {});
    // The head goes with the player as it is made; the rest comes by message.
    expect(feed.opening()[0]).toHaveLength(96000);
    feed.attach(player.node);
    expect(samples()).toHaveLength(0);
    await drain();
    const sent = samples();
    expect(sent).toHaveLength(2 * 48000);
    expect(sent.every((value, i) => value === PRE_SKIP + 96000 + i)).toBe(true);
    read(48000);
    await drain();
    expect(samples()).toHaveLength(3 * 48000);
    expect(samples()[3 * 48000 - 1]).toBe(PRE_SKIP + 5 * 48000 - 1);
  });

  it('keeps ahead where the decoder answers each packet 20 ms late, as on a busy page', async () => {
    const copy = fakeCopy(10);
    serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    await file.ready();
    stubDecoder(20);
    vi.useFakeTimers();
    try {
      const player = fakeNode();
      feedOf(file, { from: 0, until: copy.frames, loop: null }, player);
      // Packet by packet that took 130 answers, 2.6 s: longer than the 2 s head lasts (CI task 651).
      await vi.advanceTimersByTimeAsync(200);
      expect(player.samples()).toHaveLength(4 * 48000);
      expect(player.samples().every((value, i) => value === PRE_SKIP + i)).toBe(true);
    } finally {
      vi.useRealTimers();
    }
  });

  it('jumps in mid-way after settling the decoder on what comes before', async () => {
    const copy = fakeCopy(10);
    serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    await file.ready();
    const player = fakeNode();
    const { samples } = player;
    const from = 6 * 48000 + 123;
    feedOf(file, { from, until: copy.frames, loop: null }, player);
    await drain();
    expect(samples()[0]).toBe(PRE_SKIP + from);
    expect(samples().every((value, i) => value === PRE_SKIP + from + i)).toBe(true);
    const fed = decoders.at(-1)!.fed;
    expect(fed[0]).toBeLessThanOrEqual(PRE_SKIP + from - 32 * 960);
    expect(fed[0]).toBeGreaterThan(PRE_SKIP + from - 34 * 960);
  });

  it('goes round its loop to the sample and says when it has played out once', async () => {
    const copy = fakeCopy(3);
    serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    await file.ready();
    const looped = fakeNode();
    const start = 100000;
    feedOf(file, { from: 0, until: copy.frames, loop: { start, end: copy.frames } }, looped);
    await drain();
    const sent = looped.samples();
    expect(sent[copy.frames - 1]).toBe(PRE_SKIP + copy.frames - 1);
    expect(sent[copy.frames]).toBe(PRE_SKIP + start);
    const round = copy.frames - start;
    expect(sent.length).toBeGreaterThan(copy.frames + round);
    expect(sent.slice(copy.frames).every((value, i) => value === PRE_SKIP + start + (i % round))).toBe(true);

    const once = fakeNode();
    let ended = false;
    feedOf(file, { from: 0, until: copy.frames, loop: null }, once, () => (ended = true));
    await drain();
    expect(once.samples()).toHaveLength(copy.frames);
    expect(once.posted.at(-1)).toBe('end');
    once.node.port.onmessage?.({ data: 'ended' } as never);
    expect(ended).toBe(true);
  });

  it('stops the player and its decoder', async () => {
    const copy = fakeCopy(10);
    serveCopies({ [URL]: copy.bytes });
    const file = new StreamFile(URL, copy.frames);
    await file.ready();
    const player = fakeNode();
    const { posted } = player;
    const feed = feedOf(file, { from: 5 * 48000, until: copy.frames, loop: null }, player);
    await drain();
    feed.stop();
    expect(posted.at(-1)).toBe('stop');
    expect(decoders.at(-1)!.closed).toBe(true);
  });
});