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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! Getting packets back out: pages in, packets out, several logical bitstreams//! at once.//!//! This is where the two kinds of multiplexing RFC 3533 Section 4 describes//! stop needing to be told apart. Both are just pages keyed by serial number://! "grouping" interleaves pages of concurrent streams, "chaining" puts one//! whole stream after another, and a demuxer that looks up the serial number//! and reassembles per stream handles either, or the two together, without//! knowing which it is looking at.//!//! **A packet has no bound, so this is the one part of the library that//! allocates.** A page is at most 65307 bytes by construction; a packet is//! whatever a stream says it is, and a stream that ends every page mid-packet//! asks for a packet without end. `Options.max_packet_size` is the answer, and//! `Options.max_streams` is the same answer to a stream that names a new serial//! number on every page.//!//! In the ordinary case -- a packet that begins and ends inside one page --//! nothing is copied: the packet points into the page reader's buffer. A packet//! spanning pages is reassembled into a buffer belonging to its logical//! bitstream. Either way the bytes are valid only until the next call to//! `next`.//!//! ```zig//! var buffer: [ogg.Page.max_size]u8 = undefined;//! var d: ogg.Demuxer = .init(gpa, input, &buffer, .{});//! defer d.deinit();//!//! while (try d.next()) |packet| {//! // packet.serial_number says which logical bitstream it came from//! }//! ```
const std = @import("std");const assert = std.debug.assert;const Allocator = std.mem.Allocator;
const Page = @import("page.zig").Page;const PageReader = @import("reader.zig").PageReader;
/// One packet, as the codec that wrote it would recognise it.pub const Packet = struct { /// Which logical bitstream it belongs to. serial_number: u32, /// The packet. Valid until the next call to `Demuxer.next`. data: []const u8, /// Where the codec's position counter stood once this packet was decoded, /// when the stream said. Only the last packet to finish on a page gets one, /// because a page carries one granule position and that is whose it is /// (RFC 3533, Section 6, field 4). Null on every other packet, and on the /// header packets of mappings that write `Page.no_granule` for them. granule_position: ?i64, /// Counts packets within the logical bitstream from zero. index: u64, /// The first packet of its logical bitstream, off a bos page. By convention /// its first bytes identify the codec -- see `codec.identify`. first: bool, /// The last packet of its logical bitstream: it finished on the page /// carrying the end-of-stream flag. last: bool, /// Something was lost immediately before this packet -- pages missing from /// the sequence, a stream joined in the middle, or a page contradicting the /// one before it about whether a packet was still open. The packet itself is /// whole; what came before it is not. after_hole: bool,};
pub const Demuxer = struct { pub const Options = struct { /// The largest packet that will be reassembled across pages. Nothing in /// the format bounds a packet, so something has to: a stream whose every /// page ends mid-packet is asking for memory without end, and the /// honest answer is `error.PacketTooLarge`. Eight megabytes is far more /// than any codec in use needs -- Vorbis' setup header, the largest /// packet a normal file contains, runs to a few kilobytes. max_packet_size: usize = 8 << 20,
/// How many logical bitstreams may be open at once. A stream is /// forgotten once its end-of-stream page has been read, so this bounds /// how many are interleaved rather than how many a chain holds. max_streams: usize = 64, };
pub const Error = PageReader.Error || Allocator.Error || error{ /// A packet grew past `Options.max_packet_size`. PacketTooLarge, /// More logical bitstreams were open at once than /// `Options.max_streams`. TooManyStreams, };
gpa: Allocator, pages: PageReader, options: Options, streams: std.ArrayList(Stream) = .empty, /// The page being walked through, if there is one. pending: ?Pending = null, /// The reassembly buffer holding the packet handed out last, which is only /// safe to reuse once the caller has come back for another. spent: ?usize = null, /// How many pages the sequence numbers say went missing. lost_pages: u64 = 0,
const Stream = struct { serial_number: u32, /// The head of a packet that the last page did not finish. partial: std.ArrayList(u8) = .empty, /// What this stream's next page should be numbered. next_sequence: u32 = 0, /// How many packets have been handed out from it. packets: u64 = 0, /// Whether any page of it has been seen yet. seen_page: bool = false, /// Whether the first page seen was its bos page. A stream joined in the /// middle has no first packet to report. from_bos: bool = false, /// Something is missing before the next packet this stream produces. hole: bool = false, /// Its end-of-stream page has been read. ended: bool = false, };
/// Where the walk through the current page has got to. const Pending = struct { stream: usize, iterator: Page.PacketIterator, granule_position: i64, /// How many packets still finish on this page. The granule position /// belongs to the one that takes this to zero. completions_left: usize, last_page: bool, };
pub fn init( gpa: Allocator, input: *std.Io.Reader, buffer: *[Page.max_size]u8, options: Options, ) Demuxer { return .{ .gpa = gpa, .pages = .init(input, buffer), .options = options, }; }
pub fn deinit(d: *Demuxer) void { for (d.streams.items) |*stream| stream.partial.deinit(d.gpa); d.streams.deinit(d.gpa); d.* = undefined; }
/// The next packet from any logical bitstream, in the order the pages /// carrying them appear, or null at the end of the physical bitstream. pub fn next(d: *Demuxer) Error!?Packet { // Whatever was handed out last may have been living in a reassembly // buffer. The caller has come back, so it is finished with. if (d.spent) |index| { d.streams.items[index].partial.clearRetainingCapacity(); d.spent = null; }
while (true) { if (d.pending) |*pending| { if (pending.iterator.next()) |fragment| { if (try d.consume(pending, fragment)) |packet| return packet; continue; } // The page is used up. A stream whose end it was is forgotten // here, which is what keeps a long chain from counting against // `max_streams`. const index = pending.stream; d.pending = null; if (d.streams.items[index].ended) d.retire(index); continue; }
const page = (try d.pages.next()) orelse return null; try d.begin(page); } }
/// How many logical bitstreams are open. A stream is counted from its first /// page until its last has been walked through. pub fn openStreams(d: *const Demuxer) usize { return d.streams.items.len; }
/// Start walking through a page: find the logical bitstream it belongs to, /// or start one, and notice anything the sequence number says is missing. fn begin(d: *Demuxer, page: Page) Error!void { const index = d.find(page.serial_number) orelse index: { if (d.streams.items.len == d.options.max_streams) return error.TooManyStreams; try d.streams.append(d.gpa, .{ .serial_number = page.serial_number, .from_bos = page.header_type.first, // A stream met in the middle has already lost everything before // the page in hand, which is exactly what `hole` reports. .hole = !page.header_type.first, }); break :index d.streams.items.len - 1; }; const stream = &d.streams.items[index];
if (page.header_type.first and stream.seen_page) { // A bos page for a serial number already in use. Chained streams // are required to have distinct serial numbers (RFC 3533, // Section 4) and this one has not, but starting over costs nothing // and losing the rest of the file would. stream.partial.clearRetainingCapacity(); stream.packets = 0; stream.from_bos = true; stream.ended = false; stream.hole = false; } else if (stream.seen_page and page.sequence_number != stream.next_sequence) { // "a 4 Byte field containing the sequence number of the page so the // decoder can identify page loss" -- so identify it: whatever this // stream was in the middle of is gone. d.lost_pages += page.sequence_number -% stream.next_sequence; stream.partial.clearRetainingCapacity(); stream.hole = true; }
stream.seen_page = true; stream.next_sequence = page.sequence_number +% 1; if (page.header_type.last) stream.ended = true;
d.pending = .{ .stream = index, .iterator = page.packets(), .granule_position = page.granule_position, .completions_left = page.completedPackets(), .last_page = page.header_type.last, }; }
/// Take one fragment out of the page being walked through, and hand back a /// packet if it finished one. fn consume(d: *Demuxer, pending: *Pending, fragment: Page.Fragment) Error!?Packet { const stream = &d.streams.items[pending.stream]; const mid_packet = stream.partial.items.len != 0;
if (fragment.continued and !mid_packet) { // The tail of a packet whose head we never had. Nothing can be done // with it but say so. stream.hole = true; if (fragment.complete) pending.completions_left -= 1; return null; } if (!fragment.continued and mid_packet) { // The previous page ended inside a packet and this one says it is // starting a fresh one. One of them is wrong; the half packet is // what has to go. stream.partial.clearRetainingCapacity(); stream.hole = true; }
if (!fragment.complete) { try d.append(stream, fragment.bytes); return null; }
pending.completions_left -= 1;
// Nothing is copied when the packet arrived whole in this page. const data = if (stream.partial.items.len == 0) fragment.bytes else data: { try d.append(stream, fragment.bytes); d.spent = pending.stream; break :data stream.partial.items; };
const packet: Packet = .{ .serial_number = stream.serial_number, .data = data, .granule_position = if (pending.completions_left == 0 and pending.granule_position != Page.no_granule) pending.granule_position else null, .index = stream.packets, .first = stream.packets == 0 and stream.from_bos, .last = pending.last_page and fragment.final, .after_hole = stream.hole, }; stream.packets += 1; stream.hole = false; return packet; }
fn append(d: *Demuxer, stream: *Stream, bytes: []const u8) Error!void { if (stream.partial.items.len + bytes.len > d.options.max_packet_size) { return error.PacketTooLarge; } try stream.partial.appendSlice(d.gpa, bytes); }
fn find(d: *const Demuxer, serial_number: u32) ?usize { for (d.streams.items, 0..) |stream, i| { if (stream.serial_number == serial_number) return i; } return null; }
fn retire(d: *Demuxer, index: usize) void { d.streams.items[index].partial.deinit(d.gpa); _ = d.streams.swapRemove(index); }};
const testing = std.testing;const StreamWriter = @import("writer.zig").StreamWriter;
const tone_flac = @embedFile("tests/data/tone.oga");const tone_vorbis = @embedFile("tests/data/tone.ogg");const grouped = @embedFile("tests/data/grouped.ogg");
/// A demuxer over `stream`, with its buffers on the heap so that a test does/// not carry 64 kB of stack.const Fixture = struct { input: std.Io.Reader, buffer: *[Page.max_size]u8, demuxer: Demuxer,
fn init(stream: []const u8, options: Demuxer.Options) !*Fixture { const f = try testing.allocator.create(Fixture); f.* = .{ .input = .fixed(stream), .buffer = try testing.allocator.create([Page.max_size]u8), .demuxer = undefined, }; f.demuxer = .init(testing.allocator, &f.input, f.buffer, options); return f; }
fn deinit(f: *Fixture) void { f.demuxer.deinit(); testing.allocator.destroy(f.buffer); testing.allocator.destroy(f); }};
test "every packet of an Ogg FLAC stream" { const f = try Fixture.init(tone_flac, .{}); defer f.deinit();
// One identification packet, one metadata packet, six audio frames: the // page table in src/tests/data/README.md says 1 + 1 + 6. var index: u64 = 0; var last_seen = false; while (try f.demuxer.next()) |packet| : (index += 1) { try testing.expectEqual(@as(u32, 0x2c45036d), packet.serial_number); try testing.expectEqual(index, packet.index); try testing.expectEqual(index == 0, packet.first); try testing.expect(!packet.after_hole); if (index == 0) try testing.expectEqualSlices(u8, "\x7fFLAC", packet.data[0..5]); // Every FLAC audio frame starts with the format's sync code. if (index >= 2) try testing.expectEqual(@as(u8, 0xff), packet.data[0]); last_seen = packet.last; } try testing.expectEqual(@as(u64, 8), index); try testing.expect(last_seen);}
test "the granule position lands on the last packet to finish on its page" { const f = try Fixture.init(tone_flac, .{}); defer f.deinit();
var positions: std.ArrayList(?i64) = .empty; defer positions.deinit(testing.allocator); while (try f.demuxer.next()) |packet| try positions.append(testing.allocator, packet.granule_position);
// The first two pages hold one packet each and say granule zero, which is // what the Ogg FLAC mapping requires of a header page. The last page holds // six audio frames and says 6000 -- the whole of the 0.75 s at 8 kHz -- so // only the sixth of them gets a position. try testing.expectEqualSlices(?i64, &.{ 0, 0, null, null, null, null, null, 6000 }, positions.items);}
test "two logical bitstreams interleaved page by page" { const f = try Fixture.init(grouped, .{}); defer f.deinit();
// A Vorbis stream and an Opus stream, muxed into one file: both bos pages // come first, then their pages alternate. var vorbis: u64 = 0; var opus: u64 = 0; var serials: [2]u32 = .{ 0, 0 }; while (try f.demuxer.next()) |packet| { if (packet.first) { if (std.mem.startsWith(u8, packet.data, "\x01vorbis")) serials[0] = packet.serial_number; if (std.mem.startsWith(u8, packet.data, "OpusHead")) serials[1] = packet.serial_number; } if (packet.serial_number == serials[0]) vorbis += 1 else opus += 1; try testing.expect(!packet.after_hole); } try testing.expect(serials[0] != 0 and serials[1] != 0); try testing.expect(serials[0] != serials[1]); try testing.expectEqual(@as(u64, 28), vorbis); try testing.expectEqual(@as(u64, 40), opus); // Both ends were reached, so both streams were forgotten. try testing.expectEqual(@as(usize, 0), f.demuxer.openStreams());}
test "one stream chained after another" { // Concatenating two physical bitstreams is all chaining is (RFC 3533, // Section 4), so an Ogg Vorbis file followed by an Ogg FLAC file is a legal // chain of two. const f = try Fixture.init(tone_vorbis ++ tone_flac, .{}); defer f.deinit();
var firsts: std.ArrayList(u32) = .empty; defer firsts.deinit(testing.allocator); var lasts: usize = 0; var total: usize = 0; while (try f.demuxer.next()) |packet| { if (packet.first) try firsts.append(testing.allocator, packet.serial_number); if (packet.last) lasts += 1; total += 1; try testing.expect(!packet.after_hole); } try testing.expectEqualSlices(u32, &.{ 0x784d00be, 0x2c45036d }, firsts.items); try testing.expectEqual(@as(usize, 2), lasts); try testing.expectEqual(@as(usize, 28 + 8), total); // Never more than one at a time, and none left at the end. try testing.expectEqual(@as(usize, 0), f.demuxer.openStreams());}
/// Write `packets` out as one logical bitstream, read them back, and check they/// are the packets that went in. `target_body_size` decides how much of the/// spanning machinery gets exercised.fn roundTrip(packets: []const []const u8, target_body_size: usize) !void { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); { const body = try testing.allocator.create([Page.max_body_size]u8); defer testing.allocator.destroy(body); var w: StreamWriter = .init(&out.writer, body, .{ .serial_number = 0xfeedface, .target_body_size = target_body_size, }); for (packets, 0..) |packet, i| try w.writePacket(packet, @intCast(i)); try w.finish(); }
const f = try Fixture.init(out.written(), .{}); defer f.deinit();
var index: usize = 0; while (try f.demuxer.next()) |packet| : (index += 1) { try testing.expect(index < packets.len); try testing.expectEqualSlices(u8, packets[index], packet.data); try testing.expectEqual(@as(u32, 0xfeedface), packet.serial_number); try testing.expectEqual(index == 0, packet.first); try testing.expectEqual(index == packets.len - 1, packet.last); try testing.expect(!packet.after_hole); } try testing.expectEqual(packets.len, index);}
test "a round trip gives back the packets that went in" { var arena_state: std.heap.ArenaAllocator = .init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
// The lengths worth naming: nothing, one byte, either side of a segment, // either side of a page, and one packet longer than any page can hold. const lengths = [_]usize{ 0, 1, 254, 255, 256, 509, 510, 511, 65024, 65025, 65026, 200000 }; var packets: std.ArrayList([]const u8) = .empty; for (lengths) |length| { const bytes = try arena.alloc(u8, length); for (bytes, 0..) |*byte, i| byte.* = @truncate(i *% 31 +% length); try packets.append(arena, bytes); }
// Pages of the default size, pages as small as the format allows, and pages // as large as it allows. for ([_]usize{ 4096, 1, Page.max_body_size }) |target| { try roundTrip(packets.items, target); }}
test "a packet that never ends is refused rather than obeyed" { // A page whose segment table is 255 lacing values of 255, repeated: every // page continues the same packet and none of them finishes it. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
const lacing = @as([Page.max_segments]u8, @splat(255)); const body = try testing.allocator.alloc(u8, Page.max_body_size); defer testing.allocator.free(body); @memset(body, 0xa5);
for (0..64) |i| { const page: Page = .{ .granule_position = Page.no_granule, .serial_number = 1, .sequence_number = @intCast(i), .checksum = 0, .header_type = .{ .first = i == 0, .continued = i != 0 }, .lacing = &lacing, .body = body, }; try page.write(&out.writer); }
const f = try Fixture.init(out.written(), .{ .max_packet_size = 256 * 1024 }); defer f.deinit(); try testing.expectError(error.PacketTooLarge, f.demuxer.next());}
test "a stream that names a new serial number on every page is refused" { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); for (0..40) |i| { const page: Page = .{ .granule_position = Page.no_granule, .serial_number = @intCast(i + 1), .sequence_number = 0, .checksum = 0, // No end-of-stream page, so none of them is ever forgotten. .header_type = .{ .first = true }, .lacing = &.{255}, .body = &@as([255]u8, @splat(0)), }; try page.write(&out.writer); }
const f = try Fixture.init(out.written(), .{ .max_streams = 8 }); defer f.deinit(); try testing.expectError(error.TooManyStreams, f.demuxer.next());}
test "a missing page is a hole, and what follows it still arrives" { // Take the last page out of the middle of a chain of three single-packet // pages by writing the stream with a sequence number skipped. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); const bodies = [_][]const u8{ "first", "second", "third" }; for (bodies, 0..) |body, i| { const page: Page = .{ .granule_position = @intCast(i), .serial_number = 3, .sequence_number = if (i == 2) 5 else @intCast(i), .checksum = 0, .header_type = .{ .first = i == 0, .last = i == bodies.len - 1 }, .lacing = &.{@intCast(body.len)}, .body = body, }; try page.write(&out.writer); }
const f = try Fixture.init(out.written(), .{}); defer f.deinit();
try testing.expect(!(try f.demuxer.next()).?.after_hole); try testing.expect(!(try f.demuxer.next()).?.after_hole); const third = (try f.demuxer.next()).?; try testing.expectEqualSlices(u8, "third", third.data); try testing.expect(third.after_hole); try testing.expectEqual(@as(u64, 3), f.demuxer.lost_pages);}
test "the tail of a packet whose head was lost is thrown away" { // A stream joined in the middle: its first page continues a packet nobody // has the start of, and then carries a whole one. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); const page: Page = .{ .granule_position = 99, .serial_number = 4, .sequence_number = 12, .checksum = 0, .header_type = .{ .continued = true }, .lacing = &.{ 4, 5 }, .body = "tailwhole", }; try page.write(&out.writer);
const f = try Fixture.init(out.written(), .{}); defer f.deinit();
const only = (try f.demuxer.next()).?; try testing.expectEqualSlices(u8, "whole", only.data); try testing.expect(only.after_hole); // It is the last packet to finish on the page, so the position is its. try testing.expectEqual(@as(?i64, 99), only.granule_position); // Joining in the middle means there is no first packet to report. try testing.expect(!only.first); try testing.expectEqual(@as(?Packet, null), try f.demuxer.next());}
test "a page that contradicts the one before it loses the half packet" { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); const first: Page = .{ .granule_position = Page.no_granule, .serial_number = 5, .sequence_number = 0, .checksum = 0, .header_type = .{ .first = true }, // Ends mid-packet: the last lacing value is 255. .lacing = &.{255}, .body = &@as([255]u8, @splat(0x11)), }; try first.write(&out.writer); const second: Page = .{ .granule_position = 1, .serial_number = 5, .sequence_number = 1, .checksum = 0, // ...but this page says it is starting a fresh packet. .header_type = .{ .last = true }, .lacing = &.{3}, .body = "new", }; try second.write(&out.writer);
const f = try Fixture.init(out.written(), .{}); defer f.deinit(); const only = (try f.demuxer.next()).?; try testing.expectEqualSlices(u8, "new", only.data); try testing.expect(only.after_hole); try testing.expect(only.last); try testing.expectEqual(@as(?Packet, null), try f.demuxer.next());}
test "an empty physical bitstream has no packets" { const f = try Fixture.init("", .{}); defer f.deinit(); try testing.expectEqual(@as(?Packet, null), try f.demuxer.next());}