diff --git a/pkg/blob/prefix_reader.go b/pkg/blob/prefix_reader.go new file mode 100644 index 00000000..bd623416 --- /dev/null +++ b/pkg/blob/prefix_reader.go @@ -0,0 +1,84 @@ +package blob + +import ( + "errors" + "io" +) + +// PrefixReader presents header bytes followed by a byte-range window of an +// underlying blob.Reader as a single blob.Reader (io.ReaderAt + Size + Close). +// +// It's the serving primitive for flat-MP4 VOD: a synthesized faststart MP4 +// header concatenated with the canonical MUXL blob's body bytes. muxl's +// flat-header synthesis owns all absolute offsets — the moov's co64 entries +// already point into the [header][body] address space — so we serve exactly +// header ++ body[bodyOffset : bodyOffset+bodyLen] and hand it to the same Range +// machinery the raw blob uses. For a whole VOD the window is the entire blob; +// for a clip it's the contiguous sub-range muxl sized the header for. +type PrefixReader struct { + header []byte + body Reader + bodyOffset int64 // start of the window within body + bodyLen int64 // length of the window +} + +// NewPrefixReader builds a Reader over header ++ body[bodyOffset:bodyOffset+bodyLen]. +// It takes ownership of body: Close closes it. +func NewPrefixReader(header []byte, body Reader, bodyOffset, bodyLen int64) *PrefixReader { + return &PrefixReader{header: header, body: body, bodyOffset: bodyOffset, bodyLen: bodyLen} +} + +func (r *PrefixReader) Size() int64 { return int64(len(r.header)) + r.bodyLen } + +func (r *PrefixReader) Close() error { return r.body.Close() } + +// ReadAt implements io.ReaderAt over the virtual [header][body-window] file. +func (r *PrefixReader) ReadAt(p []byte, off int64) (int, error) { + if off < 0 { + return 0, errors.New("blob.PrefixReader: negative offset") + } + total := r.Size() + if off >= total { + return 0, io.EOF + } + + // Trim the request to what's actually available; a request running past the + // end reports io.EOF once the available bytes are delivered. + eof := false + if avail := total - off; int64(len(p)) > avail { + p = p[:avail] + eof = true + } + + n := 0 + h := int64(len(r.header)) + + // Header portion. + if off < h { + hn := h - off + if hn > int64(len(p)) { + hn = int64(len(p)) + } + copy(p[:hn], r.header[off:off+hn]) + n += int(hn) + off += hn + p = p[hn:] + } + + // Body window portion. + if len(p) > 0 { + m, err := r.body.ReadAt(p, r.bodyOffset+(off-h)) + n += m + if err != nil && !errors.Is(err, io.EOF) { + return n, err + } + if errors.Is(err, io.EOF) { + return n, io.EOF + } + } + + if eof { + return n, io.EOF + } + return n, nil +} diff --git a/pkg/blob/prefix_reader_test.go b/pkg/blob/prefix_reader_test.go new file mode 100644 index 00000000..d1852111 --- /dev/null +++ b/pkg/blob/prefix_reader_test.go @@ -0,0 +1,70 @@ +package blob + +import ( + "bytes" + "io" + "testing" +) + +// bytesBlob is a blob.Reader backed by an in-memory byte slice. +type bytesBlob struct{ b []byte } + +func (r bytesBlob) ReadAt(p []byte, off int64) (int, error) { + return bytes.NewReader(r.b).ReadAt(p, off) +} +func (r bytesBlob) Size() int64 { return int64(len(r.b)) } +func (r bytesBlob) Close() error { return nil } + +func TestPrefixReader(t *testing.T) { + header := []byte("HEADER!") // 7 bytes + body := bytesBlob{[]byte("0123456789")} // 10 bytes + r := NewPrefixReader(header, body, 2, 5) // window "23456" + const want = "HEADER!23456" // 12 bytes + + if got := r.Size(); got != int64(len(want)) { + t.Fatalf("Size = %d, want %d", got, len(want)) + } + + // Full read via SectionReader (exercises ReadAt across the boundary). + all, err := io.ReadAll(io.NewSectionReader(r, 0, r.Size())) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if string(all) != want { + t.Fatalf("ReadAll = %q, want %q", all, want) + } + + cases := []struct { + name string + off, n int64 + want string + wantEOF bool + }{ + {"header only", 0, 4, "HEAD", false}, + {"header into body", 4, 6, "ER!234", false}, + {"body only", 7, 3, "234", false}, + {"exact tail", 9, 3, "456", false}, + {"past end trims + EOF", 10, 5, "56", true}, + {"start at last byte", 11, 1, "6", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + buf := make([]byte, tc.n) + got, err := r.ReadAt(buf, tc.off) + if string(buf[:got]) != tc.want { + t.Fatalf("ReadAt(%d,%d) = %q, want %q", tc.off, tc.n, buf[:got], tc.want) + } + if tc.wantEOF && err != io.EOF { + t.Fatalf("ReadAt(%d,%d) err = %v, want io.EOF", tc.off, tc.n, err) + } + if !tc.wantEOF && err != nil { + t.Fatalf("ReadAt(%d,%d) err = %v, want nil", tc.off, tc.n, err) + } + }) + } + + // Reading exactly at the end is io.EOF with no bytes. + if n, err := r.ReadAt(make([]byte, 4), r.Size()); n != 0 || err != io.EOF { + t.Fatalf("ReadAt at end = (%d,%v), want (0, EOF)", n, err) + } +}