audioPlaybackWorker function

Future<void> audioPlaybackWorker(WorkerChannel channel)

Decodes an audio track and keeps a SharedAudioRing fed until it runs out or is told to stop.

Implementation

Future<void> audioPlaybackWorker(WorkerChannel channel) async {
  final init = channel.initialMessage! as Map<String, Object?>;
  final ring = SharedAudioRing.attach(
    (init['ring']! as PlatformValue).value!,
  );

  var stopped = false;
  var eof = false;
  var framesWritten = 0;
  int? seekTargetUs;
  var seekGeneration = 0;
  int? firstPtsUs;
  Object? failure;

  PlatformDemuxer? demuxer;
  PlatformAudioDecoder? decoder;

  channel.handleRequests((Object? request) async {
    switch (request) {
      case 'stats':
        return <String, Object?>{
          'framesWritten': framesWritten,
          'firstPtsUs': firstPtsUs,
          'seekGeneration': seekGeneration,
          'eof': eof,
          'underruns': ring.underruns,
          'buffered': ring.availableFrames,
          'error': failure?.toString(),
        };
      case ['seek', final int targetUs]:
        // Recorded, not performed: the pump owns the demuxer and the decoder,
        // and seeking them from under it mid-decode is how a decoder ends up
        // emitting samples from two different places in the stream. The pump
        // picks this up at its next boundary, which it reaches promptly
        // because a pending seek also breaks it out of a full-ring wait.
        seekTargetUs = targetUs;
        return null;
      case 'stop':
        stopped = true;
        return null;
      default:
        throw StateError('unknown op: $request');
    }
  });

  try {
    demuxer = ContainerFramingBackend.openInProcess(
      init['bytes']! as Uint8List,
      null,
    );
    if (demuxer == null) {
      throw const CodecInitException(
        'audio-worker',
        'no pure-Dart parser claims this container',
      );
    }

    final trackIndex = _audioTrackIndex(demuxer.tracks);
    final track = demuxer.tracks[trackIndex] as AudioTrackInfo;
    decoder = await WebCodecsAudioDecoder.create(
      AudioDecoderConfig.fromTrack(track),
    );

    // The pump. Nothing in here touches the main thread.
    bool interrupted() => stopped || seekTargetUs != null;

    while (!stopped) {
      final target = seekTargetUs;
      if (target != null) {
        seekTargetUs = null;
        await demuxer.seek(target);
        // Drop the decoder's reference state along with the container
        // position: a decoder carried across a seek emits the tail of where it
        // used to be.
        await decoder.flush();
        // Safe here and only here: the host suspends the audio thread across a
        // seek, so there is no consumer to race, and dropping what is queued is
        // the entire point of seeking.
        ring.clear();
        firstPtsUs = null;
        eof = false;
        seekGeneration++;
        continue;
      }

      final packet = await demuxer.readPacket();
      if (packet == null) {
        eof = true;
        for (final chunk in await decoder.flush()) {
          framesWritten += await _fill(ring, chunk, interrupted);
        }
        // Not a break: a seek can restart a finished stream, and exiting here
        // would leave the worker alive but permanently deaf to one.
        while (!stopped && seekTargetUs == null) {
          await Future<void>.delayed(_kRingFullPatience);
        }
        continue;
      }
      if (packet.trackIndex != trackIndex) continue;
      for (final chunk in await decoder.decode(packet)) {
        firstPtsUs ??= chunk.ptsUs;
        framesWritten += await _fill(ring, chunk, interrupted);
        if (interrupted()) break;
      }
    }
  } on Object catch (e) {
    // Recorded rather than thrown: the host asks for stats and can report a
    // dead pump, where an unhandled error in a worker is just a worker that
    // stopped for no stated reason.
    failure = e;
  }

  await channel.onClose;
  try {
    await decoder?.close();
    await demuxer?.close();
  } on Object {
    // Teardown is best effort; the host is already gone.
  }
}