Something went wrong. Try again.
Reads and writes the Ogg container format in Zig: pages, packets, grouping and chaining, with a page layer that allocates nothing.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! Taking pages off a stream, one at a time.//!//! This is the layer between `std.Io.Reader` and `Page`: it finds where each//! page ends, which needs its segment table read first, and hands back a `Page`//! borrowing the bytes. It allocates nothing -- a page is at most//! `Page.max_size` bytes, so one buffer of that size is enough forever, and the//! caller supplies it.//!//! **A failed read consumes nothing.** When the bytes at the current position//! are not a page, or are a page whose checksum is wrong, the reader is left//! exactly where it was. That is what makes `resync` possible: it can step over//! one byte and look for the next capture pattern, which is the whole reason//! the format has one (RFC 3533, Section 6, field 1).
const std = @import("std");const assert = std.debug.assert;
const Page = @import("page.zig").Page;
pub const PageReader = struct { input: *std.Io.Reader, /// Where pages are assembled. Every `Page` handed out borrows from here and /// is invalidated by the next call to `next` or `resync`. buffer: *[Page.max_size]u8, /// The page being looked at starts here. start: usize = 0, /// Nothing beyond here has been read from `input` yet. end: usize = 0,
/// How many pages `next` has returned. pages: u64 = 0, /// How many bytes `resync` has thrown away, which is a measure of how /// damaged the stream is. dropped: u64 = 0,
/// `error.EndOfStream` is not in here: a stream that ends where a page /// could begin has simply ended, and one that ends anywhere else was /// truncated. pub const Error = error{ReadFailed} || Page.ParseError;
pub fn init(input: *std.Io.Reader, buffer: *[Page.max_size]u8) PageReader { return .{ .input = input, .buffer = buffer }; }
/// The next page, or null at the end of the stream. /// /// The page borrows this reader's buffer, so it is worth reading whatever /// is wanted out of it -- or copying it -- before asking for another. /// /// Anything other than a whole, checksummed page where one was expected is /// an error, and leaves the stream positioned at the byte the trouble /// started at so that `resync` can go looking for the next page. pub fn next(pr: *PageReader) Error!?Page { pr.fill(Page.min_header_size) catch |err| switch (err) { // A stream that ends on a page boundary ends here, which is the // ordinary way for one to end. error.EndOfStream => return if (pr.start == pr.end) null else error.Truncated, error.ReadFailed => return error.ReadFailed, };
// The capture pattern and the version are checked before the byte // count is believed, so that reading a stream of something else does // not turn into waiting for 65 kB of it. const header = pr.buffer[pr.start..pr.end]; if (!std.mem.eql(u8, header[0..4], &Page.capture_pattern)) return error.NotAPage; if (header[4] != Page.supported_version) return error.UnsupportedVersion;
const segments = header[26]; try pr.fillPage(Page.min_header_size + @as(usize, segments)); const lacing = pr.buffer[pr.start + Page.min_header_size ..][0..segments]; const total = Page.min_header_size + @as(usize, segments) + Page.bodySize(lacing); try pr.fillPage(total);
const page = try Page.parse(pr.buffer[pr.start..][0..total]); pr.start += total; pr.pages += 1; return page; }
/// Throw bytes away until a page turns up whose checksum is right, and /// return it; null if the stream ends first. `dropped` says how much was /// given up on. /// /// This is the recovery path, for a caller reading a stream that may be /// damaged or that it joined in the middle of. It steps over at least one /// byte, so calling it after a failed `next` always makes progress. pub fn resync(pr: *PageReader) std.Io.Reader.Error!?Page { while (true) { pr.fill(Page.min_header_size) catch |err| switch (err) { error.EndOfStream => { // What is left is too short to be a page, so there is no // page left to find. pr.dropped += pr.end - pr.start; pr.start = pr.end; return null; }, error.ReadFailed => return error.ReadFailed, };
const window = pr.buffer[pr.start..pr.end]; if (std.mem.find(u8, window, &Page.capture_pattern)) |at| { pr.start += at; pr.dropped += at; if (pr.next()) |found| return found.? else |err| switch (err) { // Not a page after all, or not an undamaged one. Step over // the pattern that misled us and carry on looking. error.NotAPage, error.UnsupportedVersion, error.Truncated, error.ChecksumMismatch => { pr.start += 1; pr.dropped += 1; }, error.ReadFailed => return error.ReadFailed, } } else { // Nothing in what is buffered. The last three bytes could // still be the beginning of a capture pattern, so they stay. const keep = @min(window.len, Page.capture_pattern.len - 1); pr.dropped += window.len - keep; pr.start = pr.end - keep; } } }
/// Ensure `n` bytes are buffered, treating the end of the stream as a /// truncated page rather than as the end of the stream: by the time this /// is called a page header has already said how long the page is. fn fillPage(pr: *PageReader, n: usize) Error!void { pr.fill(n) catch |err| switch (err) { error.EndOfStream => return error.Truncated, error.ReadFailed => return error.ReadFailed, }; }
fn fill(pr: *PageReader, n: usize) std.Io.Reader.Error!void { assert(n <= Page.max_size); if (pr.end - pr.start >= n) return; if (pr.start > 0) { // Slide what is left to the front. The buffer is exactly one page // long, so there is nowhere else for the rest of a page to go. std.mem.copyForwards(u8, pr.buffer[0 .. pr.end - pr.start], pr.buffer[pr.start..pr.end]); pr.end -= pr.start; pr.start = 0; } while (pr.end - pr.start < n) { const got = try pr.input.readSliceShort(pr.buffer[pr.end..]); if (got == 0) return error.EndOfStream; pr.end += got; } }};
const testing = std.testing;
const tone_flac = @embedFile("tests/data/tone.oga");const tone_vorbis = @embedFile("tests/data/tone.ogg");
/// The page sizes and flags of src/tests/data/tone.oga, read off the file with/// a script of its own: a bos page holding the Ogg FLAC identification packet,/// a page holding the Vorbis comment block the mapping requires next, and a/// page of audio which is also the end of the stream.const tone_flac_pages = [_]struct { size: usize, flags: u8, granule: i64, packets: usize }{ .{ .size = 79, .flags = 0x02, .granule = 0, .packets = 1 }, .{ .size = 81, .flags = 0x00, .granule = 0, .packets = 1 }, .{ .size = 2519, .flags = 0x04, .granule = 6000, .packets = 6 },};
test "every page of a stream, in order" { var input: std.Io.Reader = .fixed(tone_flac); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer);
for (tone_flac_pages, 0..) |expected, i| { const page = (try pr.next()).?; try testing.expectEqual(expected.size, page.size()); try testing.expectEqual(expected.flags, @as(u8, @bitCast(page.header_type))); try testing.expectEqual(expected.granule, page.granule_position); try testing.expectEqual(expected.packets, page.completedPackets()); try testing.expectEqual(@as(u32, @intCast(i)), page.sequence_number); try testing.expectEqual(@as(u32, 0x2c45036d), page.serial_number); } try testing.expectEqual(@as(?Page, null), try pr.next()); try testing.expectEqual(@as(u64, 3), pr.pages); try testing.expectEqual(@as(u64, 0), pr.dropped);}
test "reading pages back out of one byte at a time" { // A `Reader` with a one-byte buffer, to prove nothing here depends on how // much the layer underneath hands over at once. var fixed: std.Io.Reader = .fixed(tone_vorbis); var one: [1]u8 = undefined; var limited = fixed.limited(.unlimited, &one); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&limited.interface, &buffer);
var count: usize = 0; var total: usize = 0; while (try pr.next()) |page| { count += 1; total += page.size(); } try testing.expectEqual(@as(usize, 3), count); try testing.expectEqual(tone_vorbis.len, total);}
test "a truncated last page is an error, not a short page" { var input: std.Io.Reader = .fixed(tone_flac[0 .. tone_flac.len - 1]); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer); _ = try pr.next(); _ = try pr.next(); try testing.expectError(error.Truncated, pr.next());}
test "trailing bytes that are not a page are an error" { const stream = tone_flac ++ "junk"; var input: std.Io.Reader = .fixed(stream); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer); for (0..3) |_| _ = try pr.next(); try testing.expectError(error.Truncated, pr.next());}
test "resync steps over damage and finds the next page" { // Garbage in front of the stream, and a bit flipped in the middle of the // second page, which leaves one good page at each end of the mess. var stream: [64 + tone_flac.len]u8 = undefined; for (stream[0..64], 0..) |*byte, i| byte.* = @truncate(i *% 7); @memcpy(stream[64..], tone_flac); stream[64 + 79 + 40] ^= 0x80;
var input: std.Io.Reader = .fixed(&stream); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer);
// Nothing at the front is a page. try testing.expectError(error.NotAPage, pr.next());
const first = (try pr.resync()).?; try testing.expectEqual(@as(u32, 0), first.sequence_number); try testing.expectEqual(@as(u64, 64), pr.dropped);
// The second page is the damaged one, so `next` refuses it and `resync` // walks on to the third. try testing.expectError(error.ChecksumMismatch, pr.next()); const third = (try pr.resync()).?; try testing.expectEqual(@as(u32, 2), third.sequence_number); try testing.expectEqual(@as(u64, 64 + 81), pr.dropped);
try testing.expectEqual(@as(?Page, null), try pr.resync());}
test "resync on a stream holding no page at all gives up" { var stream: [200000]u8 = undefined; for (&stream, 0..) |*byte, i| byte.* = @truncate(i *% 31); var input: std.Io.Reader = .fixed(&stream); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer); try testing.expectEqual(@as(?Page, null), try pr.resync()); try testing.expectEqual(@as(u64, stream.len), pr.dropped);}
test "a capture pattern straddling two reads is still found" { // The pattern is split across the point where the reader has to go back // for more bytes, which is the case the three kept bytes are for. var stream: [Page.max_size * 2]u8 = undefined; const pad = Page.max_size - 2; for (stream[0..pad], 0..) |*byte, i| byte.* = @truncate(i *% 13); @memcpy(stream[pad..][0..tone_flac.len], tone_flac);
var input: std.Io.Reader = .fixed(stream[0 .. pad + tone_flac.len]); var buffer: [Page.max_size]u8 = undefined; var pr: PageReader = .init(&input, &buffer); const page = (try pr.resync()).?; try testing.expectEqual(@as(u32, 0x2c45036d), page.serial_number); try testing.expectEqual(@as(u64, pad), pr.dropped);}