Something went wrong. Try again.
A pure Zig PipeWire client library that speaks the native wire protocol to play audio and manage the graph, with no libpipewire and no libc.
Something went wrong. Try again.
13 kB · 350 lines
Zig
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! The PipeWire native protocol connection: framing, buffering and descriptor//! passing over the daemon's Unix socket.//!//! Each message is a 16-byte header of four native-endian `u32`s —//!//! [0] id target object id//! [1] opcode in the high 8 bits, payload size in the low 24//! [2] seq sequence number, echoed back by Sync/Done//! [3] n_fds how many of the connection's pending descriptors are this//! message's//!//! — followed by the payload, which is a single POD (usually a Struct), and//! optionally a second "footer" POD that this library does not send and ignores//! on receipt. Descriptors travel out of band as SCM_RIGHTS, and a message//! refers to them by index rather than by number.
const std = @import("std");const sys = @import("sys.zig");const pod = @import("pod.zig");
pub const header_size = 16;pub const max_payload = 0xffffff;
/// The sequence counter wraps within this mask, matching `SPA_ASYNC_SEQ_MASK`.pub const seq_mask: u32 = 0x3fffffff;
pub const Error = sys.Error || std.mem.Allocator.Error || error{ProtocolError};
/// One received message. `data` and `fds` borrow the connection's buffers and/// stay valid until `release` is called or another message is read.pub const Message = struct { id: u32, opcode: u8, seq: u32, data: []const u8, fds: []i32,
/// A parser over the message payload. The payload is a single POD; any /// footer after it is left alone. pub fn parser(m: Message) pod.Parser { return pod.Parser.init(m.data); }
/// Take ownership of the descriptor at `index`, so that `release` will not /// close it. Returns -1 if the index is out of range or already taken. pub fn takeFd(m: Message, index: i64) i32 { if (index < 0 or index >= m.fds.len) return -1; const i: usize = @intCast(index); const fd = m.fds[i]; m.fds[i] = -1; return fd; }};
pub const Connection = struct { gpa: std.mem.Allocator, fd: i32,
in: std.ArrayListAligned(u8, .@"8") = .empty, /// How much of `in` has been handed out as complete messages. in_pos: usize = 0, in_fds: std.ArrayList(i32) = .empty, /// How many of `in_fds` belong to messages already handed out. in_fds_pos: usize = 0,
out: std.ArrayList(u8) = .empty, out_fds: std.ArrayList(i32) = .empty,
seq: u32 = 0,
pub fn init(gpa: std.mem.Allocator, fd: i32) Connection { return .{ .gpa = gpa, .fd = fd }; }
pub fn deinit(c: *Connection) void { for (c.in_fds.items[c.in_fds_pos..]) |fd| sys.close(fd); c.in.deinit(c.gpa); c.in_fds.deinit(c.gpa); c.out.deinit(c.gpa); c.out_fds.deinit(c.gpa); sys.close(c.fd); c.fd = -1; }
// --- sending ---
/// Queue a message. `fds` are duplicated into the outgoing queue by index, /// so `payload` must already refer to them by the indices `addFd` returned. pub fn send(c: *Connection, id: u32, opcode: u8, payload: []const u8, n_fds: u32) Error!u32 { if (payload.len > max_payload) return error.ProtocolError;
const seq = c.seq; c.seq = (c.seq + 1) & seq_mask;
var hdr: [4]u32 = .{ id, (@as(u32, opcode) << 24) | @as(u32, @intCast(payload.len)), seq, n_fds, }; try c.out.appendSlice(c.gpa, std.mem.sliceAsBytes(hdr[0..])); try c.out.appendSlice(c.gpa, payload); return seq; }
/// Register a descriptor for the message being built and return the index to /// put in its Fd POD. The descriptor is closed once the message is flushed. pub fn addFd(c: *Connection, fd: i32) Error!i64 { const index = c.out_fds.items.len; try c.out_fds.append(c.gpa, fd); return @intCast(index); }
/// Write everything queued to the socket. /// /// Descriptors must accompany the message that owns them, so a batch is cut /// short at `max_fds_per_msg`; the C implementation does the same. pub fn flush(c: *Connection) Error!void { var data: []const u8 = c.out.items; var fds: []const i32 = c.out_fds.items; var closed: usize = 0;
defer { // Whatever went out is dropped from the queues; anything left stays // for the next flush. for (c.out_fds.items[0..closed]) |fd| sys.close(fd); const remaining_fds = c.out_fds.items.len - closed; std.mem.copyForwards(i32, c.out_fds.items[0..remaining_fds], c.out_fds.items[closed..]); c.out_fds.shrinkRetainingCapacity(remaining_fds);
const sent = c.out.items.len - data.len; const remaining = c.out.items.len - sent; std.mem.copyForwards(u8, c.out.items[0..remaining], c.out.items[sent..]); c.out.shrinkRetainingCapacity(remaining); }
while (data.len > 0) { const batch_fds = @min(fds.len, sys.max_fds_per_msg); // When more descriptors are queued than one message may carry, send // a token amount of payload with them and let the rest follow. const batch_len = if (fds.len > sys.max_fds_per_msg) @min(data.len, @sizeOf(u32)) else data.len;
const n = sys.sendmsgFds(c.fd, data[0..batch_len], fds[0..batch_fds]) catch |err| switch (err) { error.Interrupted => continue, else => return err, }; data = data[n..]; fds = fds[batch_fds..]; closed += batch_fds; } }
// --- receiving ---
/// Read whatever the socket has available into the input buffer. /// /// With `nonblocking` set this returns `error.Again` when nothing is ready. pub fn fill(c: *Connection, nonblocking: bool) Error!void { c.compactInput(); // Grow by a page at a time; PipeWire's own buffer starts at 32 KiB. const chunk = 32 * 1024; try c.in.ensureUnusedCapacity(c.gpa, chunk); try c.in_fds.ensureUnusedCapacity(c.gpa, sys.max_fds_per_msg);
const spare = c.in.unusedCapacitySlice(); const fd_spare = c.in_fds.unusedCapacitySlice();
while (true) { const r = sys.recvmsgFds(c.fd, spare, fd_spare, nonblocking) catch |err| switch (err) { error.Interrupted => continue, else => return err, }; c.in.items.len += r.len; c.in_fds.items.len += r.n_fds; return; } }
/// Drop already-consumed bytes and descriptors from the front of the input /// buffers so they do not grow without bound. fn compactInput(c: *Connection) void { if (c.in_pos > 0) { const remaining = c.in.items.len - c.in_pos; std.mem.copyForwards(u8, c.in.items[0..remaining], c.in.items[c.in_pos..]); c.in.shrinkRetainingCapacity(remaining); c.in_pos = 0; } if (c.in_fds_pos > 0) { const remaining = c.in_fds.items.len - c.in_fds_pos; std.mem.copyForwards(i32, c.in_fds.items[0..remaining], c.in_fds.items[c.in_fds_pos..]); c.in_fds.shrinkRetainingCapacity(remaining); c.in_fds_pos = 0; } }
/// The next complete message in the buffer, or null if more bytes are needed. /// /// The caller must `release` each message before asking for the next one. pub fn next(c: *Connection) Error!?Message { const avail = c.in.items[c.in_pos..]; if (avail.len < header_size) return null;
const words = std.mem.bytesAsSlice(u32, avail[0..header_size]); const id = words[0]; const opcode: u8 = @intCast(words[1] >> 24); const size: usize = words[1] & 0xffffff; const seq = words[2]; const n_fds: usize = words[3];
if (avail.len < header_size + size) return null;
const fds_avail = c.in_fds.items.len - c.in_fds_pos; if (n_fds > fds_avail) { // The payload arrived before its descriptors; wait for another read. return null; }
const msg: Message = .{ .id = id, .opcode = opcode, .seq = seq, .data = avail[header_size..][0..size], .fds = c.in_fds.items[c.in_fds_pos..][0..n_fds], }; c.in_pos += header_size + size; c.in_fds_pos += n_fds; return msg; }
/// Close any descriptors of `msg` the handler did not take. pub fn release(c: *Connection, msg: Message) void { _ = c; for (msg.fds) |fd| sys.close(fd); }};
// --- Tests -----------------------------------------------------------------
const testing = std.testing;
test "framing round-trips over a socketpair" { var fds: [2]i32 = undefined; const rc = std.os.linux.socketpair(std.os.linux.AF.UNIX, std.os.linux.SOCK.STREAM, 0, &fds); try testing.expectEqual(@as(usize, 0), rc);
var a = Connection.init(testing.allocator, fds[0]); defer a.deinit(); var b = Connection.init(testing.allocator, fds[1]); defer b.deinit();
var builder = pod.Builder.init(testing.allocator); defer builder.deinit(); const f = try builder.pushStruct(); try builder.addInt(3); try builder.addString("hi"); try builder.pop(f);
_ = try a.send(7, 5, builder.bytes(), 0); try a.flush();
try b.fill(false); const msg = (try b.next()) orelse return error.TestUnexpectedResult; defer b.release(msg);
try testing.expectEqual(@as(u32, 7), msg.id); try testing.expectEqual(@as(u8, 5), msg.opcode); try testing.expectEqual(@as(usize, 0), msg.fds.len);
var p = msg.parser(); var fields = try p.nextStruct(); try testing.expectEqual(@as(i32, 3), try fields.nextInt()); try testing.expectEqualStrings("hi", try fields.nextString());}
test "descriptors arrive with their message" { var fds: [2]i32 = undefined; _ = std.os.linux.socketpair(std.os.linux.AF.UNIX, std.os.linux.SOCK.STREAM, 0, &fds);
var a = Connection.init(testing.allocator, fds[0]); defer a.deinit(); var b = Connection.init(testing.allocator, fds[1]); defer b.deinit();
const ev = try sys.eventfd(0, 0); const index = try a.addFd(ev); try testing.expectEqual(@as(i64, 0), index);
var builder = pod.Builder.init(testing.allocator); defer builder.deinit(); const f = try builder.pushStruct(); try builder.addFd(index); try builder.pop(f);
_ = try a.send(0, 1, builder.bytes(), 1); try a.flush();
try b.fill(false); const msg = (try b.next()) orelse return error.TestUnexpectedResult; defer b.release(msg); try testing.expectEqual(@as(usize, 1), msg.fds.len);
var p = msg.parser(); var fields = try p.nextStruct(); const got = msg.takeFd(try fields.nextFd()); try testing.expect(got >= 0); // The descriptor is a working eventfd on this side of the socket too. try sys.eventfdWrite(got, 5); try testing.expectEqual(@as(u64, 5), try sys.eventfdRead(got)); sys.close(got); // Taking it means release must not close it a second time. try testing.expectEqual(@as(i32, -1), msg.fds[0]);}
test "a message split across reads is only returned once whole" { var fds: [2]i32 = undefined; _ = std.os.linux.socketpair(std.os.linux.AF.UNIX, std.os.linux.SOCK.STREAM, 0, &fds);
var a = Connection.init(testing.allocator, fds[0]); defer a.deinit(); var b = Connection.init(testing.allocator, fds[1]); defer b.deinit();
var builder = pod.Builder.init(testing.allocator); defer builder.deinit(); const f = try builder.pushStruct(); try builder.addString("a somewhat longer payload to split"); try builder.pop(f); _ = try a.send(1, 2, builder.bytes(), 0);
// Hand the receiver the header and a few bytes, then the rest. const all = a.out.items; _ = try sys.sendmsgFds(a.fd, all[0 .. header_size + 4], &.{}); try b.fill(false); try testing.expect(try b.next() == null);
_ = try sys.sendmsgFds(a.fd, all[header_size + 4 ..], &.{}); a.out.clearRetainingCapacity(); try b.fill(false); const msg = (try b.next()) orelse return error.TestUnexpectedResult; defer b.release(msg); var p = msg.parser(); var fields = try p.nextStruct(); try testing.expectEqualStrings("a somewhat longer payload to split", try fields.nextString());}