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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! Putting packets into pages: one logical bitstream at a time.//!//! A codec hands over packets and Ogg has to decide where pages end. That is//! the only real decision in writing the format, and it is a trade: a page is//! the unit of error recovery and of seeking, so small pages make a stream//! robust, and each one costs 27 bytes of header plus a lacing value per 255//! bytes of content. RFC 3533 says pages are "usually 4-8 kB" (Section 6) and//! `Options.target_body_size` starts there.//!//! Like the reader, this allocates nothing: a page body is at most//! `Page.max_body_size` bytes, so one buffer of that size is enough forever.//!//! ```zig//! var body: [ogg.Page.max_body_size]u8 = undefined;//! var w: ogg.StreamWriter = .init(output, &body, .{ .serial_number = serial });//!//! try w.writePacket(identification_header, 0); // alone on the bos page//! try w.writePacket(comment_header, 0);//! for (audio) |packet| try w.writePacket(packet.bytes, packet.samples_so_far);//! try w.finish();//! ```//!//! # Several logical bitstreams in one physical one//!//! Interleaving two streams -- "grouping", RFC 3533 Section 4 -- is done by//! pointing two `StreamWriter`s at the same `std.Io.Writer`. The rule the//! format imposes is that every bos page comes before any other page, so write//! and `flush` each stream's identification packet first, and only then start//! handing over content. Chaining, where one stream's eos page is followed by//! the next one's bos page, needs nothing at all: `finish` one and start the//! next.
const std = @import("std");const assert = std.debug.assert;
const Page = @import("page.zig").Page;const HeaderType = @import("page.zig").HeaderType;
pub const StreamWriter = struct { pub const Options = struct { /// Names the logical bitstream. Every page of it carries this, and it /// has to be different from every other stream's in the same physical /// bitstream -- including ones chained before or after, which is why /// the format asks for a random number rather than a counter. serial_number: u32,
/// A page is closed once its body has reached this much -- see /// `writePacket` for exactly when the bytes go out. Larger pages cost /// proportionally less header; smaller ones lose less to a bad sector /// and let a seek land closer. The hard bound is `Page.max_body_size`, /// which a value above is clamped to by the segment table running out. target_body_size: usize = 4096,
/// Close the page after the first packet, so that the identification /// packet sits alone on the bos page. Every media mapping in use /// requires that -- it is how a reader can identify the codec from the /// first page alone -- so leave it on unless writing something that /// does not. flush_after_first_packet: bool = true, };
pub const Error = std.Io.Writer.Error;
output: *std.Io.Writer, /// Where the page being built keeps its content. body: *[Page.max_body_size]u8, /// The page being built. 255 entries is the most a segment table holds. lacing: [Page.max_segments]u8 = undefined,
serial_number: u32, target_body_size: usize, flush_after_first_packet: bool,
sequence_number: u32 = 0, /// Bytes handed to the output, header and body of every page written. bytes: u64 = 0, segments: usize = 0, body_len: usize = 0,
/// The granule position of the last packet to have finished in the page /// being built, or null if none has. A page where no packet finishes /// carries `Page.no_granule`. granule: ?i64 = null, /// The last granule position written on any page, so that an end-of-stream /// page with nothing on it still says where the stream got to. last_granule: i64 = Page.no_granule, /// The page being built opens with the tail of a packet. continued: bool = false, /// Whether any page has been written, which is what makes the next one the /// bos page or not. started: bool = false, /// The page being built has reached `target_body_size` and goes out as soon /// as there is something to follow it. See `writePacket`. flush_pending: bool = false, /// Whether `finish` has been called. ended: bool = false, /// `writeLastPacket` is in progress, so no page goes out until `finish` /// can put the end-of-stream flag on it. ending: bool = false,
pub fn init(output: *std.Io.Writer, body: *[Page.max_body_size]u8, options: Options) StreamWriter { return .{ .output = output, .body = body, .serial_number = options.serial_number, .target_body_size = options.target_body_size, .flush_after_first_packet = options.flush_after_first_packet, }; }
/// Add one packet, splitting it over as many pages as it needs. /// /// `granule_position` is what the codec's position counter stands at once /// this packet has been decoded, in whatever unit the media mapping chose. /// It lands on the page the packet finishes in, and a page where nothing /// finishes gets `Page.no_granule` instead -- which is the whole of what /// the field means (RFC 3533, Section 6, field 4). A packet whose position /// is not known, which is what the header packets of most mappings are, /// takes `Page.no_granule`. /// /// A packet of no bytes is not an error; the format has a lacing value for /// it. /// /// **A page that has reached `target_body_size` goes out when the next /// packet arrives, not when it fills.** Holding it back is what lets /// `finish` put the end-of-stream flag on a page with content on it rather /// than on an empty page after it. A caller who needs the bytes out sooner /// than that -- anything writing a live stream -- calls `flush`. pub fn writePacket(w: *StreamWriter, packet: []const u8, granule_position: i64) Error!void { assert(!w.ended); if (w.flush_pending and w.segments > 0) try w.emit(false); const was_first = !w.started and w.segments == 0;
var rest = packet; while (true) { // A full segment table has to go out before anything else can be // added, and the packet then continues on the page after it. if (w.segments == Page.max_segments) try w.emit(false);
const n = @min(rest.len, 255); @memcpy(w.body[w.body_len..][0..n], rest[0..n]); w.lacing[w.segments] = @intCast(n); w.segments += 1; w.body_len += n; rest = rest[n..];
// A lacing value below 255 ends the packet -- which is why a packet // whose length is a multiple of 255 needs a trailing zero, and gets // one from the next turn round this loop. if (n < 255) break; } w.granule = granule_position;
// The bos page has to go out now rather than when the next packet // arrives: in a physical bitstream carrying several logical ones, every // bos page has to precede every other page, and a writer that held this // one back would let another stream's content overtake it. // // `sequence_number` is still zero only if the packet fitted in the bos // page; a first packet long enough to have filled one already has not // got a page to itself whatever is done here. if (w.ending) return; if (was_first and w.flush_after_first_packet and w.sequence_number == 0) return w.flush(); if (w.body_len >= w.target_body_size) w.flush_pending = true; }
/// Add the stream's last packet and close the stream, so that the /// end-of-stream flag lands on the page carrying it. /// /// This is worth using wherever the caller knows which packet is the last /// one. `writePacket` followed by `finish` reaches the same place, but if /// the packet happened to close a page on its way in -- which it always /// does for a stream of one packet, since the identification packet goes /// alone on the bos page -- the flag has nowhere to go but an empty page /// after it. pub fn writeLastPacket(w: *StreamWriter, packet: []const u8, granule_position: i64) Error!void { assert(!w.ended); w.ending = true; try w.writePacket(packet, granule_position); try w.finish(); }
/// Close the page being built, if anything is in it. What follows starts a /// new one. pub fn flush(w: *StreamWriter) Error!void { assert(!w.ended); if (w.segments == 0) return; try w.emit(false); }
/// Close the stream: write what is pending with the end-of-stream flag set. /// /// If nothing is pending -- the last packet happened to fill a page exactly, /// or was followed by a `flush` -- the flag goes on a page of its own with /// no content, which RFC 3533 Section 4 explicitly allows and which still /// reports the position the stream reached. pub fn finish(w: *StreamWriter) Error!void { assert(!w.ended); if (w.segments == 0) w.granule = w.last_granule; try w.emit(true); w.ended = true; }
/// Bytes of this logical bitstream handed to the output so far. A page /// still being built is not counted, since none of it has been written. pub fn bytesWritten(w: *const StreamWriter) u64 { return w.bytes; }
fn emit(w: *StreamWriter, last: bool) Error!void { const page: Page = .{ .granule_position = w.granule orelse Page.no_granule, .serial_number = w.serial_number, .sequence_number = w.sequence_number, .checksum = 0, // `Page.write` computes the real one. .header_type = .{ .continued = w.continued, .first = !w.started, .last = last, }, .lacing = w.lacing[0..w.segments], .body = w.body[0..w.body_len], }; try page.write(w.output);
w.bytes += page.size(); w.started = true; w.sequence_number +%= 1; if (w.granule) |granule| w.last_granule = granule; // The page that follows continues a packet exactly when this one ended // inside one. w.continued = w.segments > 0 and w.lacing[w.segments - 1] == 255; w.segments = 0; w.body_len = 0; w.granule = null; w.flush_pending = false; }};
const testing = std.testing;const PageReader = @import("reader.zig").PageReader;
/// Write `packets` as one logical bitstream and give back the bytes.fn writeStream(packets: []const []const u8, options: StreamWriter.Options) ![]u8 { var out: std.Io.Writer.Allocating = .init(testing.allocator); errdefer 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, options); for (packets, 0..) |packet, i| try w.writePacket(packet, @intCast(i)); try w.finish(); return out.toOwnedSlice();}
/// Every page of `stream`, so a test can say what the framing came out as.fn pagesOf(stream: []const u8, out: *std.ArrayList(Page)) !void { var input: std.Io.Reader = .fixed(stream); const buffer = try testing.allocator.create([Page.max_size]u8); defer testing.allocator.destroy(buffer); var pr: PageReader = .init(&input, buffer); // Every page is copied out with its own body, since the reader hands back // slices of one buffer it reuses. while (try pr.next()) |page| { try out.append(testing.allocator, .{ .granule_position = page.granule_position, .serial_number = page.serial_number, .sequence_number = page.sequence_number, .checksum = page.checksum, .header_type = page.header_type, .lacing = try testing.allocator.dupe(u8, page.lacing), .body = try testing.allocator.dupe(u8, page.body), }); }}
fn freePages(pages: *std.ArrayList(Page)) void { for (pages.items) |page| { testing.allocator.free(page.lacing); testing.allocator.free(page.body); } pages.deinit(testing.allocator);}
test "the first packet goes alone on the bos page" { const stream = try writeStream(&.{ "identification", "comment", "audio" }, .{ .serial_number = 7 }); defer testing.allocator.free(stream);
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(stream, &pages);
try testing.expectEqual(@as(usize, 2), pages.items.len); try testing.expect(pages.items[0].header_type.first); try testing.expectEqualSlices(u8, "identification", pages.items[0].body); try testing.expectEqual(@as(u32, 7), pages.items[0].serial_number);
try testing.expect(!pages.items[1].header_type.first); try testing.expect(pages.items[1].header_type.last); try testing.expectEqualSlices(u8, "commentaudio", pages.items[1].body); try testing.expectEqualSlices(u8, &.{ 7, 5 }, pages.items[1].lacing); try testing.expectEqual(@as(u32, 1), pages.items[1].sequence_number);}
test "the granule position on a page is the last one to finish on it" { const stream = try writeStream(&.{ "a", "b", "c", "d" }, .{ .serial_number = 1, .flush_after_first_packet = false, }); defer testing.allocator.free(stream);
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(stream, &pages);
// All four fit in one page, and `writeStream` numbers them 0, 1, 2, 3. try testing.expectEqual(@as(usize, 1), pages.items.len); try testing.expectEqual(@as(i64, 3), pages.items[0].granule_position);}
test "a packet longer than a page spans pages, and only the last carries a position" { // 255 segments is a page's worth of table, so a packet of 255 * 255 bytes // fills one exactly and its terminating zero has to go on the next. const long = try testing.allocator.alloc(u8, Page.max_body_size); defer testing.allocator.free(long); for (long, 0..) |*byte, i| byte.* = @truncate(i *% 11);
const stream = try writeStream(&.{long}, .{ .serial_number = 1 }); defer testing.allocator.free(stream);
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(stream, &pages);
try testing.expectEqual(@as(usize, 2), pages.items.len); try testing.expect(pages.items[0].header_type.first); try testing.expect(pages.items[0].endsMidPacket()); try testing.expectEqual(@as(i64, Page.no_granule), pages.items[0].granule_position); try testing.expectEqual(@as(usize, Page.max_body_size), pages.items[0].body.len);
try testing.expect(pages.items[1].header_type.continued); try testing.expect(pages.items[1].header_type.last); try testing.expectEqualSlices(u8, &.{0}, pages.items[1].lacing); try testing.expectEqual(@as(i64, 0), pages.items[1].granule_position);}
test "a packet of no bytes is a lacing value of zero" { const stream = try writeStream(&.{ "x", "", "" }, .{ .serial_number = 1, .flush_after_first_packet = false, }); defer testing.allocator.free(stream);
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(stream, &pages); try testing.expectEqualSlices(u8, &.{ 1, 0, 0 }, pages.items[0].lacing); try testing.expectEqual(@as(usize, 1), pages.items[0].body.len);}
test "a stream with no packets at all is one nil page" { 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 = 3 }); try w.finish();
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(out.written(), &pages);
try testing.expectEqual(@as(usize, 1), pages.items.len); try testing.expect(pages.items[0].header_type.first); try testing.expect(pages.items[0].header_type.last); try testing.expectEqual(@as(usize, 0), pages.items[0].lacing.len); try testing.expectEqual(@as(usize, Page.min_header_size), pages.items[0].size());}
test "two streams in one physical bitstream, interleaved page by page" { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); const body_a = try testing.allocator.create([Page.max_body_size]u8); defer testing.allocator.destroy(body_a); const body_b = try testing.allocator.create([Page.max_body_size]u8); defer testing.allocator.destroy(body_b);
var a: StreamWriter = .init(&out.writer, body_a, .{ .serial_number = 100 }); var b: StreamWriter = .init(&out.writer, body_b, .{ .serial_number = 200 });
// Every bos page before any other page, which is what grouping requires // and which is why the writer does not hold those two back. try a.writePacket("A header", Page.no_granule); try b.writePacket("B header", Page.no_granule); try a.writePacket("A content", 1); try a.finish(); try b.writePacket("B content", 1); try b.finish();
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(out.written(), &pages);
const serials = [_]u32{ 100, 200, 100, 200 }; try testing.expectEqual(serials.len, pages.items.len); for (pages.items, serials, 0..) |page, serial, i| { try testing.expectEqual(serial, page.serial_number); try testing.expectEqual(i < 2, page.header_type.first); try testing.expectEqual(i >= 2, page.header_type.last); } // Each stream numbers its own pages. try testing.expectEqual(@as(u32, 1), pages.items[2].sequence_number); try testing.expectEqual(@as(u32, 1), pages.items[3].sequence_number);}
test "a stream of one packet is one page, opening and closing the stream" { 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 = 9 }); try w.writeLastPacket("everything", 42);
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(out.written(), &pages);
try testing.expectEqual(@as(usize, 1), pages.items.len); try testing.expect(pages.items[0].header_type.first); try testing.expect(pages.items[0].header_type.last); try testing.expectEqualSlices(u8, "everything", pages.items[0].body); try testing.expectEqual(@as(i64, 42), pages.items[0].granule_position);}
test "closing a stream whose last page has already gone out costs a nil page" { // The same one-packet stream, ended the other way. Nothing is lost -- the // end-of-stream flag and the position are both on the page that follows -- // but it is 27 bytes that `writeLastPacket` does not spend. 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 = 9 }); try w.writePacket("everything", 42); try w.finish();
var pages: std.ArrayList(Page) = .empty; defer freePages(&pages); try pagesOf(out.written(), &pages);
try testing.expectEqual(@as(usize, 2), pages.items.len); try testing.expect(pages.items[0].header_type.first); try testing.expect(!pages.items[0].header_type.last); try testing.expectEqual(@as(usize, 0), pages.items[1].lacing.len); try testing.expect(pages.items[1].header_type.last); try testing.expectEqual(@as(i64, 42), pages.items[1].granule_position);}
test "the byte count is what reached the output" { 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 = 1 }); try testing.expectEqual(@as(u64, 0), w.bytesWritten()); try w.writePacket("head", 0); // The bos page has gone out; nothing of the next page has. try testing.expectEqual(@as(u64, Page.min_header_size + 1 + 4), w.bytesWritten()); try w.writePacket("pending", 1); try testing.expectEqual(@as(u64, Page.min_header_size + 1 + 4), w.bytesWritten()); try w.finish(); try testing.expectEqual(out.written().len, w.bytesWritten());}
test "small pages cost more header and nothing else" { const forty = "0123456789" ++ "0123456789" ++ "0123456789" ++ "0123456789"; const packets = [_][]const u8{ forty, forty, forty }; const big = try writeStream(&packets, .{ .serial_number = 1, .target_body_size = 4096 }); defer testing.allocator.free(big); const small = try writeStream(&packets, .{ .serial_number = 1, .target_body_size = 1 }); defer testing.allocator.free(small); try testing.expect(small.len > big.len);
// ...and the content is the same either way, which is what the demuxer // tests check in full. var small_pages: std.ArrayList(Page) = .empty; defer freePages(&small_pages); try pagesOf(small, &small_pages); try testing.expectEqual(@as(usize, 3), small_pages.items.len);}