feat: update examplees

apps/playground/package.json

@@ -9,7 +9,7 @@
     "preview": "rsbuild preview"
   },
   "dependencies": {
-    "konoebml": "0.1.2-rc.5",
+    "konoebml": "0.1.2",
     "lit": "^3.2.1"
   },
   "devDependencies": {

apps/playground/src/media/base/errors.ts

@@ -1,4 +1,4 @@
-export class UnsupportCodecError extends Error {
+export class UnsupportedCodecError extends Error {
   constructor(codec: string, context: string) {
     super(`codec ${codec} is not supported in ${context} context`);
   }
@@ -9,3 +9,17 @@ export class ParseCodecPrivateError extends Error {
     super(`code ${codec} private parse failed: ${detail}`);
   }
 }
+
+export class UnreachableOrLogicError extends Error {
+  constructor(detail: string) {
+    super(`unreachable or logic error: ${detail}`)
+  }
+}
+
+export class ParseCodecErrors extends Error {
+  cause: Error[] = [];
+
+  constructor() {
+    super('failed to parse codecs');
+  }
+}

apps/playground/src/media/mkv/codecs/index.ts

@@ -1,5 +1,5 @@
 import { AudioCodec } from '../../base/audio_codecs';
-import { UnsupportCodecError } from '../../base/errors';
+import { UnsupportedCodecError } from '../../base/errors';
 import { VideoCodec } from '../../base/video_codecs';
 import type { TrackEntryType } from '../schema';
 import {
@@ -10,6 +10,7 @@ import {
   genCodecIdByAVCDecoderConfigurationRecord,
   parseAVCDecoderConfigurationRecord,
 } from './avc';
+import type {ProbeInfo} from "@/media/mkv/enhance/probe.ts";
 
 export const VideoCodecId = {
   VCM: 'V_MS/VFW/FOURCC',
@@ -107,123 +108,183 @@ export type SubtitleCodecIdType =
   | `${(typeof SubtitleCodecId)[keyof typeof SubtitleCodecId]}`
   | string;
 
-export function videoCodecIdToWebCodecsVideoDecoder(
-  track: TrackEntryType
-): [VideoCodec, string] {
+
+export interface VideoDecoderConfigExt extends VideoDecoderConfig {
+  codecType: VideoCodec,
+}
+
+export function videoCodecIdToWebCodecs(
+  track: TrackEntryType,
+  _probeInfo?: ProbeInfo
+): VideoDecoderConfigExt {
   const codecId = track.CodecID;
   const codecPrivate = track.CodecPrivate;
+  const shareOptions = {
+    description: codecPrivate
+  }
   switch (codecId) {
     case VideoCodecId.HEVC:
-      return [VideoCodec.HEVC, 'hevc'];
+      return { ...shareOptions, codecType: VideoCodec.HEVC, codec: 'hevc' };
     case VideoCodecId.VP9:
-      return [VideoCodec.VP9, 'vp09'];
+      return { ...shareOptions, codecType: VideoCodec.VP9, codec: 'vp09' };
     case VideoCodecId.AV1:
-      return [VideoCodec.AV1, 'av1'];
+      return { ...shareOptions, codecType: VideoCodec.AV1, codec: 'av1' };
     case VideoCodecId.H264:
       if (!codecPrivate) {
-        throw new UnsupportCodecError(
+        throw new UnsupportedCodecError(
           'h264(without codec_private profile)',
           'web codecs audio decoder'
         );
       }
-      return [
-        VideoCodec.H264,
-        genCodecIdByAVCDecoderConfigurationRecord(
+      return {
+        ...shareOptions,
+        codecType: VideoCodec.H264,
+        codec: genCodecIdByAVCDecoderConfigurationRecord(
           parseAVCDecoderConfigurationRecord(codecPrivate)
-        ),
-      ];
+         )
+      };
     case VideoCodecId.THEORA:
-      return [VideoCodec.Theora, 'theora'];
+      return { ...shareOptions, codecType: VideoCodec.Theora, codec: 'theora' };
     case VideoCodecId.VP8:
-      return [VideoCodec.VP8, 'vp8'];
+      return { ...shareOptions, codecType: VideoCodec.VP8, codec: 'vp8' };
     case VideoCodecId.MPEG4_ISO_SP:
-      return [VideoCodec.MPEG4, 'mp4v.01.3'];
+      return { ...shareOptions, codecType: VideoCodec.MPEG4, codec: 'mp4v.01.3' };
     case VideoCodecId.MPEG4_ISO_ASP:
-      return [VideoCodec.MPEG4, 'mp4v.20.9'];
+      return { ...shareOptions, codecType: VideoCodec.MPEG4, codec: 'mp4v.20.9' };
     case VideoCodecId.MPEG4_ISO_AP:
-      return [VideoCodec.MPEG4, 'mp4v.20.9'];
+      return { ...shareOptions, codecType: VideoCodec.MPEG4, codec: 'mp4v.20.9' };
     default:
-      throw new UnsupportCodecError(codecId, 'web codecs video decoder');
+      throw new UnsupportedCodecError(codecId, 'web codecs video decoder');
   }
 }
 
-export function videoCodecIdToWebCodecsAudioDecoder(
-  track: TrackEntryType
-): [AudioCodec, string] {
+export interface AudioDecoderConfigExt extends AudioDecoderConfig {
+  codecType: AudioCodec,
+}
+
+export function audioCodecIdToWebCodecs(
+  track: TrackEntryType,
+  _probeInfo?: ProbeInfo
+): AudioDecoderConfigExt {
   const codecId = track.CodecID;
   const codecPrivate = track.CodecPrivate;
   const bitDepth = track.Audio?.BitDepth;
+  const numberOfChannels = Number(track.Audio?.Channels);
+  const sampleRate = Number(track.Audio?.SamplingFrequency);
+
+  const shareOptions = {
+    numberOfChannels,
+    sampleRate,
+    description: codecPrivate
+  }
+
   switch (track.CodecID) {
     case AudioCodecId.AAC_MPEG4_MAIN:
     case AudioCodecId.AAC_MPEG2_MAIN:
-      return [AudioCodec.AAC, 'mp4a.40.1'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: 'mp4a.40.1'
+      };
     case AudioCodecId.AAC_MPEG2_LC:
     case AudioCodecId.AAC_MPEG4_LC:
-      return [AudioCodec.AAC, 'mp4a.40.2'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: 'mp4a.40.2'
+      };
     case AudioCodecId.AAC_MPEG2_SSR:
     case AudioCodecId.AAC_MPEG4_SSR:
-      return [AudioCodec.AAC, 'mp4a.40.3'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: 'mp4a.40.3'
+      };
     case AudioCodecId.AAC_MPEG4_LTP:
-      return [AudioCodec.AAC, 'mp4a.40.4'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: 'mp4a.40.4'
+      };
     case AudioCodecId.AAC_MPEG2_LC_SBR:
     case AudioCodecId.AAC_MPEG4_SBR:
-      return [AudioCodec.AAC, 'mp4a.40.5'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: 'mp4a.40.5'
+      };
     case AudioCodecId.AAC:
-      return [
-        AudioCodec.AAC,
-        codecPrivate
-          ? genCodecIdByAudioSpecificConfig(
-              parseAudioSpecificConfig(codecPrivate)
-            )
-          : 'mp4a.40.2',
-      ];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AAC,
+        codec: codecPrivate
+        ? genCodecIdByAudioSpecificConfig(
+          parseAudioSpecificConfig(codecPrivate)
+        ) : 'mp4a.40.2',
+      };
     case AudioCodecId.AC3:
     case AudioCodecId.AC3_BSID9:
-      return [AudioCodec.AC3, 'ac-3'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.AC3,
+        codec: 'ac-3'
+      };
     case AudioCodecId.EAC3:
     case AudioCodecId.AC3_BSID10:
-      return [AudioCodec.EAC3, 'ec-3'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.EAC3,
+        codec: 'ec-3'
+      };
     case AudioCodecId.MPEG_L3:
-      return [AudioCodec.MP3, 'mp3'];
+      return {
+        ...shareOptions,
+        codecType: AudioCodec.MP3,
+        codec: 'mp3'
+      };
     case AudioCodecId.VORBIS:
-      return [AudioCodec.Vorbis, 'vorbis'];
+      return { ...shareOptions, codecType: AudioCodec.Vorbis, codec: 'vorbis' }
+;
     case AudioCodecId.FLAC:
-      return [AudioCodec.FLAC, 'flac'];
+      return { ...shareOptions, codecType: AudioCodec.FLAC, codec: 'flac' }
+;
     case AudioCodecId.OPUS:
-      return [AudioCodec.Opus, 'opus'];
+      return { ...shareOptions, codecType: AudioCodec.Opus, codec: 'opus' }
+;
     case AudioCodecId.ALAC:
-      return [AudioCodec.ALAC, 'alac'];
+      return { ...shareOptions, codecType: AudioCodec.ALAC, codec: 'alac' }
+;
     case AudioCodecId.PCM_INT_BIG:
       if (bitDepth === 16) {
-        return [AudioCodec.PCM_S16BE, 'pcm-s16be'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S16BE, codec: 'pcm-s16be' };
       }
       if (bitDepth === 24) {
-        return [AudioCodec.PCM_S24BE, 'pcm-s24be'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S24BE, codec: 'pcm-s24be' };
       }
       if (bitDepth === 32) {
-        return [AudioCodec.PCM_S32BE, 'pcm-s32be'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S32BE, codec: 'pcm-s32be' };
       }
-      throw new UnsupportCodecError(
+      throw new UnsupportedCodecError(
         `${codecId}(${bitDepth}b)`,
         'web codecs audio decoder'
       );
     case AudioCodecId.PCM_INT_LIT:
       if (bitDepth === 16) {
-        return [AudioCodec.PCM_S16LE, 'pcm-s16le'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S16LE, codec: 'pcm-s16le' };
       }
       if (bitDepth === 24) {
-        return [AudioCodec.PCM_S24LE, 'pcm-s24le'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S24LE, codec: 'pcm-s24le' };
       }
       if (bitDepth === 32) {
-        return [AudioCodec.PCM_S32LE, 'pcm-s32le'];
+        return { ...shareOptions, codecType: AudioCodec.PCM_S32LE, codec: 'pcm-s32le' };
       }
-      throw new UnsupportCodecError(
+      throw new UnsupportedCodecError(
         `${codecId}(${bitDepth}b)`,
         'web codecs audio decoder'
       );
     case AudioCodecId.PCM_FLOAT_IEEE:
-      return [AudioCodec.PCM_F32LE, 'pcm-f32le'];
+      return { ...shareOptions, codecType: AudioCodec.PCM_F32LE, codec: 'pcm-f32le' };
     default:
-      throw new UnsupportCodecError(codecId, 'web codecs audio decoder');
+      throw new UnsupportedCodecError(codecId, 'web codecs audio decoder');
   }
 }

apps/playground/src/media/mkv/codecs/vp9.ts

@@ -0,0 +1,21 @@
+import { type } from 'arktype';
+import type {TrackEntryType} from "@/media/mkv/schema.ts";
+
+export const VP9DecoderProfileSchema = type('0 | 1 | 2 | 3');
+
+export const VP9DecoderConfigurationRecordSchema = type({
+  profile: VP9DecoderProfileSchema,
+  level: type.number,
+  bitDepth: type.number,
+});
+
+export type VP9DecoderConfigurationRecordType =
+  typeof VP9DecoderConfigurationRecordSchema.infer;
+
+export function parseVP9DecoderConfigurationRecord(track: TrackEntryType) {
+  const pixelWidth = Number(track.Video?.PixelWidth);
+  const pixelHeight = Number(track.Video?.PixelHeight);
+  const pixels = pixelWidth * pixelHeight;
+  const bitDepth = Number(track.Video?.Colour?.BitsPerChannel) || 10;
+
+}

apps/playground/src/media/mkv/enhance/probe.ts

@@ -0,0 +1,3 @@
+export interface ProbeInfo {
+
+}

apps/playground/src/media/mkv/model.ts

@@ -30,8 +30,14 @@ import {
   TagSchema,
   type TagType,
   TrackEntrySchema,
-  type TrackEntryType,
+  type TrackEntryType, TrackTypeRestrictionEnum,
 } from './schema';
+import {concatBufs} from "konoebml/lib/tools";
+import {ParseCodecErrors, UnreachableOrLogicError, UnsupportedCodecError} from "@/media/base/errors.ts";
+import type {ProbeInfo} from "@/media/mkv/enhance/probe.ts";
+import {audioCodecIdToWebCodecs, videoCodecIdToWebCodecs} from "@/media/mkv/codecs";
+import {Queue} from "mnemonist";
+import {BehaviorSubject} from "rxjs";
 
 export const SEEK_ID_KAX_INFO = new Uint8Array([0x15, 0x49, 0xa9, 0x66]);
 export const SEEK_ID_KAX_TRACKS = new Uint8Array([0x16, 0x54, 0xae, 0x6b]);
@@ -41,6 +47,10 @@ export const SEEK_ID_KAX_TAGS = new Uint8Array([0x12, 0x54, 0xc3, 0x67]);
 export class SegmentSystem {
   startTag: EbmlSegmentTagType;
   headTags: EbmlTagType[] = [];
+  teeStream: ReadableStream<Uint8Array>
+  teeBufferTask: Promise<Uint8Array>;
+  firstCluster: EbmlClusterTagType | undefined;
+  probInfo?: ProbeInfo;
 
   cue: CueSystem;
   cluster: ClusterSystem;
@@ -49,7 +59,7 @@ export class SegmentSystem {
   track: TrackSystem;
   tag: TagSystem;
 
-  constructor(startNode: EbmlSegmentTagType) {
+  constructor(startNode: EbmlSegmentTagType, teeStream: ReadableStream<Uint8Array>) {
     this.startTag = startNode;
     this.cue = new CueSystem(this);
     this.cluster = new ClusterSystem(this);
@@ -57,17 +67,35 @@ export class SegmentSystem {
     this.info = new InfoSystem(this);
     this.track = new TrackSystem(this);
     this.tag = new TagSystem(this);
+    this.teeStream = teeStream;
+    this.teeBufferTask = this.teeWaitingProbingData(teeStream);
+  }
+
+  private async teeWaitingProbingData (teeStream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
+    const reader = teeStream.getReader();
+    const list: Uint8Array<ArrayBufferLike>[] = [];
+    while (true) {
+      try {
+        const { done, value } = await reader.read();
+        if (done) {
+          break;
+        }
+        list.push(value);
+      } catch (e: any) {
+        if (e?.name === 'AbortError') {
+          break;
+        }
+        throw e;
+      }
+    }
+    return concatBufs(...list)
   }
 
   get contentStartOffset() {
     return this.startTag.startOffset + this.startTag.headerLength;
   }
 
-  get startOffset() {
-    return this.startTag.startOffset;
-  }
-
-  completeHeads() {
+  private seekLocal () {
     const infoTag = this.seek.seekTagBySeekId(SEEK_ID_KAX_INFO);
     const tracksTag = this.seek.seekTagBySeekId(SEEK_ID_KAX_TRACKS);
     const cuesTag = this.seek.seekTagBySeekId(SEEK_ID_KAX_CUES);
@@ -83,13 +111,11 @@ export class SegmentSystem {
       this.track.prepareTracksWithTag(tracksTag);
     }
     if (tagsTag?.id === EbmlTagIdEnum.Tags) {
-      this.tag.prepareTagsWIthTag(tagsTag);
+      this.tag.prepareTagsWithTag(tagsTag);
     }
-
-    return this;
   }
 
-  scanHead(tag: EbmlTagType) {
+  scanMeta(tag: EbmlTagType) {
     if (
       tag.id === EbmlTagIdEnum.SeekHead &&
       tag.position === EbmlTagPosition.End
@@ -98,8 +124,62 @@ export class SegmentSystem {
     }
     this.headTags.push(tag);
     this.seek.memoTag(tag);
+    if (tag.id === EbmlTagIdEnum.Cluster && !this.firstCluster) {
+      this.firstCluster = tag;
+      this.seekLocal();
+    }
     return this;
   }
+
+  async completeMeta () {
+    this.seekLocal();
+
+    await this.parseCodes();
+
+    return this;
+  }
+
+  async fetchProbeInfo (_payload: Uint8Array): Promise<ProbeInfo> {
+    // call local or remote ff-probe
+    return {}
+  }
+
+  async parseCodes () {
+    const candidates = this.track.tracks.filter(c => c.TrackType === TrackTypeRestrictionEnum.AUDIO || c.TrackType === TrackTypeRestrictionEnum.VIDEO);
+    const parseErrors = new ParseCodecErrors();
+
+    if (!this.probInfo) {
+      for (const t of candidates) {
+        try {
+          await this.track.initTrack(t, undefined)
+        } catch (e: unknown) {
+          parseErrors.cause.push(e as Error)
+        }
+      }
+      if (parseErrors.cause.length > 0) {
+        try {
+          const teeBuffer = await this.teeBufferTask;
+          this.probInfo = await this.fetchProbeInfo(teeBuffer);
+        } catch (e) {
+          parseErrors.cause.push(e as Error);
+          return;
+        }
+      } else {
+        return;
+      }
+    }
+
+    for (const t of candidates) {
+      try {
+        await this.track.initTrack(t, this.probInfo)
+      } catch (e) {
+        parseErrors.cause.push(e as Error)
+      }
+    }
+    if (parseErrors.cause.length > 0) {
+      console.error(parseErrors);
+    }
+  }
 }
 
 export type SegmentComponent<T> = T & {
@@ -157,7 +237,7 @@ export class SeekSystem extends SegmentComponentSystemTrait<
   }
 
   seekHeads: SeekHeadType[] = [];
-  offsetToTagMemo: Map<number, EbmlTagType> = new Map();
+  private offsetToTagMemo: Map<number, EbmlTagType> = new Map();
 
   memoTag(tag: EbmlTagType) {
     this.offsetToTagMemo.set(tag.startOffset, tag);
@@ -193,6 +273,13 @@ export class SeekSystem extends SegmentComponentSystemTrait<
   seekTagBySeekId(seekId: Uint8Array): EbmlTagType | undefined {
     return this.seekTagByStartOffset(this.seekOffsetBySeekId(seekId));
   }
+
+  get firstClusterOffset () {
+    if (!this.segment.firstCluster) {
+      throw new UnreachableOrLogicError("first cluster not found")
+    }
+    return this.segment.firstCluster.startOffset;
+  }
 }
 
 export class InfoSystem extends SegmentComponentSystemTrait<
@@ -228,6 +315,18 @@ export class ClusterSystem extends SegmentComponentSystemTrait<
   }
 }
 
+export interface GetTrackEntryOptions {
+  priority?: (v: SegmentComponent<TrackEntryType>) => number;
+  predicate?: (v: SegmentComponent<TrackEntryType>) => boolean;
+}
+
+
+export interface TrackState<Decoder, Config, Frame> {
+  decoder: Decoder,
+  configuration?: Config,
+  frameBuffer$: BehaviorSubject<Queue<Frame>>
+}
+
 export class TrackSystem extends SegmentComponentSystemTrait<
   EbmlTrackEntryTagType,
   typeof TrackEntrySchema
@@ -237,15 +336,14 @@ export class TrackSystem extends SegmentComponentSystemTrait<
   }
 
   tracks: SegmentComponent<TrackEntryType>[] = [];
+  videoTrackState = new WeakMap<TrackEntryType, TrackState< VideoDecoder, VideoDecoderConfig, VideoFrame>>();
+  audioTrackState = new WeakMap<TrackEntryType, TrackState<AudioDecoder, AudioDecoderConfig, AudioData>>();
 
   getTrackEntry({
     priority = (track) =>
       (Number(!!track.FlagForced) << 4) + Number(!!track.FlagDefault),
     predicate = (track) => track.FlagEnabled !== 0,
-  }: {
-    priority?: (v: SegmentComponent<TrackEntryType>) => number;
-    predicate?: (v: SegmentComponent<TrackEntryType>) => boolean;
-  }) {
+  }: GetTrackEntryOptions) {
     return this.tracks
       .filter(predicate)
       .toSorted((a, b) => priority(b) - priority(a))
@@ -258,6 +356,52 @@ export class TrackSystem extends SegmentComponentSystemTrait<
       .map((c) => this.componentFromTag(c));
     return this;
   }
+
+  async initTrack (track: TrackEntryType, probe?: ProbeInfo)  {
+    if (track.TrackType === TrackTypeRestrictionEnum.AUDIO) {
+      const configuration = audioCodecIdToWebCodecs(track, probe);
+      if (await AudioDecoder.isConfigSupported(configuration)) {
+        throw new UnsupportedCodecError(configuration.codec, 'audio decoder')
+      }
+
+      const queue$ = new BehaviorSubject(new Queue<AudioData>());
+      this.audioTrackState.set(track, {
+        configuration,
+        decoder: new AudioDecoder({
+          output: (audioData) => {
+            const queue = queue$.getValue();
+            queue.enqueue(audioData);
+            queue$.next(queue);
+          },
+          error: (e) => {
+            queue$.error(e);
+          },
+        }),
+        frameBuffer$: queue$,
+      })
+    } else if (track.TrackType === TrackTypeRestrictionEnum.VIDEO) {
+      const configuration = videoCodecIdToWebCodecs(track, probe);
+      if (await VideoDecoder.isConfigSupported(configuration)) {
+        throw new UnsupportedCodecError(configuration.codec, 'audio decoder')
+      }
+
+      const queue$ = new BehaviorSubject(new Queue<VideoFrame>());
+      this.videoTrackState.set(track, {
+        configuration,
+        decoder: new VideoDecoder({
+          output: (audioData) => {
+            const queue = queue$.getValue();
+            queue.enqueue(audioData);
+            queue$.next(queue);
+          },
+          error: (e) => {
+            queue$.error(e);
+          },
+        }),
+        frameBuffer$: queue$,
+      })
+    }
+  }
 }
 
 export class CueSystem extends SegmentComponentSystemTrait<
@@ -350,7 +494,7 @@ export class TagSystem extends SegmentComponentSystemTrait<
 
   tags: SegmentComponent<TagType>[] = [];
 
-  prepareWithTagsTag(tag: EbmlTagsTagType) {
+  prepareTagsWithTag(tag: EbmlTagsTagType) {
     this.tags = tag.children
       .filter((c) => c.id === EbmlTagIdEnum.Tag)
       .map((c) => this.componentFromTag(c));

apps/playground/src/media/mkv/reactive.ts

@@ -1,62 +1,64 @@
+import {EbmlStreamDecoder, EbmlTagIdEnum, EbmlTagPosition, type EbmlTagType,} from 'konoebml';
 import {
-  type EbmlTagType,
-  EbmlStreamDecoder,
-  EbmlTagIdEnum,
-  EbmlTagPosition,
-} from 'konoebml';
-import {
-  Observable,
-  from,
-  switchMap,
-  share,
   defer,
   EMPTY,
-  of,
   filter,
   finalize,
+  from,
   isEmpty,
   map,
   merge,
-  raceWith,
+  Observable,
+  of,
   reduce,
   scan,
+  share,
   shareReplay,
+  switchMap,
   take,
-  takeUntil,
+  takeWhile,
   withLatestFrom,
 } from 'rxjs';
-import { createRangedStream, type CreateRangedStreamOptions } from '@/fetch';
-import {
-  SegmentSystem,
-  SEEK_ID_KAX_CUES,
-  type CueSystem,
-  type SegmentComponent,
-  SEEK_ID_KAX_TAGS,
-} from './model';
-import { isTagIdPos, waitTick } from './util';
-import type { ClusterType } from './schema';
+import {createRangedStream, type CreateRangedStreamOptions} from '@/fetch';
+import {type CueSystem, SEEK_ID_KAX_CUES, SEEK_ID_KAX_TAGS, type SegmentComponent, SegmentSystem,} from './model';
+import {isTagIdPos, waitTick} from './util';
+import type {ClusterType} from './schema';
 
 export interface CreateRangedEbmlStreamOptions
-  extends CreateRangedStreamOptions {}
+  extends CreateRangedStreamOptions {
+  tee?: boolean;
+}
 
 export function createRangedEbmlStream({
   url,
   byteStart = 0,
   byteEnd,
+  tee = false
 }: CreateRangedEbmlStreamOptions): Observable<{
   ebml$: Observable<EbmlTagType>;
   totalSize?: number;
   response: Response;
   body: ReadableStream<Uint8Array>;
   controller: AbortController;
+  teeBody: ReadableStream<Uint8Array> | undefined;
 }> {
   const stream$ = from(createRangedStream({ url, byteStart, byteEnd }));
 
   return stream$.pipe(
     switchMap(({ controller, body, totalSize, response }) => {
       let requestCompleted = false;
+      let teeStream: ReadableStream<Uint8Array> | undefined;
+
+      let stream: ReadableStream<Uint8Array>;
+
+      if (tee) {
+        [stream, teeStream] = body.tee();
+      } else {
+        stream = body;
+      }
+
       const originRequest$ = new Observable<EbmlTagType>((subscriber) => {
-        body
+        stream
           .pipeThrough(
             new EbmlStreamDecoder({
               streamStartOffset: byteStart,
@@ -114,7 +116,8 @@ export function createRangedEbmlStream({
         ebml$,
         totalSize,
         response,
-        body,
+        body: stream,
+        teeBody: teeStream,
         controller,
       });
     })
@@ -128,14 +131,16 @@ export function createEbmlController({
   url,
   ...options
 }: CreateEbmlControllerOptions) {
-  const request$ = createRangedEbmlStream({
+  const metaRequest$ = createRangedEbmlStream({
     ...options,
     url,
     byteStart: 0,
+    tee: true
   });
 
-  const controller$ = request$.pipe(
-    map(({ totalSize, ebml$, response, controller }) => {
+  const controller$ = metaRequest$.pipe(
+    map(({ totalSize, ebml$, response, controller, teeBody }) => {
+
       const head$ = ebml$.pipe(
         filter(isTagIdPos(EbmlTagIdEnum.EBML, EbmlTagPosition.End)),
         take(1),
@@ -147,8 +152,7 @@ export function createEbmlController({
       );
 
       const segmentStart$ = ebml$.pipe(
-        filter((s) => s.position === EbmlTagPosition.Start),
-        filter((tag) => tag.id === EbmlTagIdEnum.Segment)
+        filter(isTagIdPos(EbmlTagIdEnum.Segment, EbmlTagPosition.Start))
       );
 
       /**
@@ -157,24 +161,24 @@ export function createEbmlController({
        */
       const segments$ = segmentStart$.pipe(
         map((startTag) => {
-          const segment = new SegmentSystem(startTag);
+          const segment = new SegmentSystem(startTag, teeBody!);
           const clusterSystem = segment.cluster;
           const seekSystem = segment.seek;
 
-          const segmentEnd$ = ebml$.pipe(
-            filter(isTagIdPos(EbmlTagIdEnum.Segment, EbmlTagPosition.End)),
-            filter((tag) => tag.id === EbmlTagIdEnum.Segment),
-            take(1)
-          );
-
-          const clusterStart$ = ebml$.pipe(
-            filter(isTagIdPos(EbmlTagIdEnum.Cluster, EbmlTagPosition.Start)),
-            take(1),
-            shareReplay(1)
-          );
-
           const meta$ = ebml$.pipe(
-            takeUntil(clusterStart$.pipe(raceWith(segmentEnd$))),
+            scan((acc, tag) => {
+              // avoid object recreation
+              acc.hasKeyframe = acc.hasKeyframe || (tag.id === EbmlTagIdEnum.SimpleBlock && tag.keyframe) || (tag.id === EbmlTagIdEnum.BlockGroup && tag.children.every(c => c.id !== EbmlTagIdEnum.ReferenceBlock));
+              acc.tag = tag;
+              return acc;
+            }, { hasKeyframe: false, tag: undefined as unknown as EbmlTagType }),
+            takeWhile(
+              ({ tag, hasKeyframe }) => {
+                return !isTagIdPos(EbmlTagIdEnum.Segment, EbmlTagPosition.End)(tag) && !(isTagIdPos(EbmlTagIdEnum.Cluster, EbmlTagPosition.End)(tag) && hasKeyframe);
+              },
+              true
+            ),
+            map(({ tag }) => tag),
             share({
               resetOnComplete: false,
               resetOnError: false,
@@ -183,8 +187,8 @@ export function createEbmlController({
           );
 
           const withMeta$ = meta$.pipe(
-            reduce((segment, meta) => segment.scanHead(meta), segment),
-            map(segment.completeHeads.bind(segment)),
+            reduce((segment, meta) => segment.scanMeta(meta), segment),
+            switchMap(() => segment.completeMeta()),
             take(1),
             shareReplay(1)
           );
@@ -231,6 +235,7 @@ export function createEbmlController({
               if (tagSystem.prepared) {
                 return EMPTY;
               }
+
               const remoteTagsTagStartOffset =
                 seekSystem.seekOffsetBySeekId(SEEK_ID_KAX_TAGS);
               if (remoteTagsTagStartOffset! >= 0) {
@@ -243,7 +248,7 @@ export function createEbmlController({
                   filter(isTagIdPos(EbmlTagIdEnum.Tags, EbmlTagPosition.End)),
                   withLatestFrom(withMeta$),
                   map(([tags, withMeta]) => {
-                    withMeta.tag.prepareWithTagsTag(tags);
+                    withMeta.tag.prepareTagsWithTag(tags);
                     return withMeta;
                   })
                 );
@@ -280,12 +285,12 @@ export function createEbmlController({
           const seekWithoutCues = (
             seekTime: number
           ): Observable<SegmentComponent<ClusterType>> => {
-            const request$ = clusterStart$.pipe(
-              switchMap((startTag) =>
+            const request$ = withMeta$.pipe(
+              switchMap(() =>
                 createRangedEbmlStream({
                   ...options,
                   url,
-                  byteStart: startTag.startOffset,
+                  byteStart: seekSystem.firstClusterOffset,
                 })
               )
             );
@@ -301,18 +306,16 @@ export function createEbmlController({
 
             return cluster$.pipe(
               scan(
-                (prev, curr) =>
-                  [prev?.[1], curr] as [
-                    SegmentComponent<ClusterType> | undefined,
-                    SegmentComponent<ClusterType> | undefined,
-                  ],
-                [undefined, undefined] as [
-                  SegmentComponent<ClusterType> | undefined,
-                  SegmentComponent<ClusterType> | undefined,
-                ]
+                (acc, curr) => {
+                  // avoid object recreation
+                  acc.prev = acc.next;
+                  acc.next = curr;
+                  return acc;
+                },
+                ({ prev: undefined as (SegmentComponent<ClusterType> | undefined), next: undefined as SegmentComponent<ClusterType> | undefined })
               ),
-              filter((c) => c[1]?.Timestamp! > seekTime),
-              map((c) => c[0] ?? c[1]!)
+              filter((c) => c.next?.Timestamp! > seekTime),
+              map((c) => c.prev ?? c.next!)
             );
           };
 
@@ -394,6 +397,6 @@ export function createEbmlController({
 
   return {
     controller$,
-    request$,
+    request$: metaRequest$,
   };
 }

apps/playground/src/video-pipeline-demo.ts

@@ -23,7 +23,6 @@ import { TrackTypeRestrictionEnum, type ClusterType } from './media/mkv/schema';
 import type { SegmentComponent } from './media/mkv/model';
 import { createRef, ref, type Ref } from 'lit/directives/ref.js';
 import { Queue } from 'mnemonist';
-import { dataViewSliceToBuf } from 'konoebml';
 
 export class VideoPipelineDemo extends LitElement {
   static styles = css``;
@@ -45,7 +44,6 @@ export class VideoPipelineDemo extends LitElement {
   videoFrameBuffer$ = new BehaviorSubject(new Queue<VideoFrame>());
   audioFrameBuffer$ = new BehaviorSubject(new Queue<AudioData>());
   pipeline$$?: Subscription;
-  bridge$$?: Subscription;
   private startTime = 0;
 
   paused$ = new BehaviorSubject<boolean>(false);

pnpm-lock.yaml

@@ -89,8 +89,8 @@ importers:
   apps/playground:
     dependencies:
       konoebml:
-        specifier: 0.1.2-rc.5
-        version: 0.1.2-rc.5(arktype@2.1.10)
+        specifier: 0.1.2
+        version: 0.1.2(arktype@2.1.10)
       lit:
         specifier: ^3.2.1
         version: 3.2.1
@@ -1885,6 +1885,15 @@ packages:
     resolution: {integrity: sha512-dcS1ul+9tmeD95T+x28/ehLgd9mENa3LsvDTtzm3vyBEO7RPptvAD+t44WVXaUjTBRcrpFeFlC8WCruUR456hw==}
     engines: {node: '>=0.10.0'}
 
+  konoebml@0.1.2:
+    resolution: {integrity: sha512-ZPibYe5KLu+fhd3ZgKiz2xyitTY24VhhG+DHw+hCUwJppaQbEJDLTkbUcWQ6A3JkSkL5RiZASNocFvqfudASNA==}
+    engines: {node: '>= 18.0.0'}
+    peerDependencies:
+      arktype: ^2.0.0
+    peerDependenciesMeta:
+      arktype:
+        optional: true
+
   konoebml@0.1.2-rc.5:
     resolution: {integrity: sha512-VsXIlsXby0OzSzLER6ERRZE+9kLkqrYUF7Wr9MKAt8qvmUc3/YStf2SdpC2gMOtCjoyxDi7bXCQPIOHziUu4nw==}
     engines: {node: '>= 18.0.0'}
@@ -4518,6 +4527,13 @@ snapshots:
 
   kind-of@6.0.3: {}
 
+  konoebml@0.1.2(arktype@2.1.10):
+    dependencies:
+      mnemonist: 0.40.3
+      type-fest: 4.37.0
+    optionalDependencies:
+      arktype: 2.1.10
+
   konoebml@0.1.2-rc.5(arktype@2.1.10):
     dependencies:
       mnemonist: 0.40.3