srctree

Andrew Kelley parent d6a8352c 472276f0
client supports sending request to queue an album

and the server understands the request
README.md added: 145, removed: 54, total 91
@@ -1,12 +1,19 @@
# player
# Groove Basin
 
A Zig package for music playback.
Music player server with a web-based user interface.
 
Everything needed to build a music player application except for the user
interface.
All clients listen to the same music at the same time, sharing access to
playback controls, the play queue, and the music library.
 
Groove Basin operates based on a directory of files. It does not stream content
from third parties. However, it integrates with third parties for various
features.
 
## Status
 
The web app is not usable yet, however there are a handful of proof-of-concept
example applications.
 
The "playlist" example plays a queue of audio files, gaplessly, over the
default output device. It also shows printing metadata tags.
 
@@ -42,8 +49,14 @@ require some porting work in the dependency tree.
zig build
```
 
## Development
 
This extension makes it easy to debug Zig WebAssembly code:
[C/C++ DevTools Support (DWARF)](https://chromewebstore.google.com/detail/cc++-devtools-support-dwa/pdcpmagijalfljmkmjngeonclgbbannb)
 
## Roadmap
 
* use the thread pool when scanning the music directory
* click an album to queue it and play it
* example that transcodes a list of input files
* update playlist example to support scripted inputs including pausing,
 
client/main.zig added: 145, removed: 54, total 91
@@ -8,6 +8,7 @@ const Db = @import("shared").Db;
const js = struct {
extern "js" fn log(ptr: [*]const u8, len: usize) void;
extern "js" fn panic(ptr: [*]const u8, len: usize) noreturn;
extern "js" fn send(ptr: [*]const u8, len: usize) void;
};
 
pub const std_options: std.Options = .{
@@ -103,3 +104,12 @@ export fn albums_len() usize {
export fn album_title(index: u32) String {
return String.init(db.optStringToSlice(db.albums.keys()[index].title) orelse "");
}
 
export fn enqueueAlbum(index: Db.Album.Index) void {
const message: protocol.EnqueueAlbum = .{ .album_index = index };
send(std.mem.asBytes(&message));
}
 
fn send(bytes: []const u8) void {
js.send(bytes.ptr, bytes.len);
}
 
server/WebSocket.zig added: 145, removed: 54, total 91
@@ -11,6 +11,8 @@ request: *std.http.Server.Request,
recv_fifo: std.fifo.LinearFifo(u8, .Slice),
reader: std.io.AnyReader,
response: std.http.Server.Response,
/// Number of bytes that have been peeked but not discarded yet.
outstanding_len: usize,
 
pub const InitError = error{WebSocketUpgradeMissingKey} ||
std.http.Server.Request.ReaderError;
@@ -65,6 +67,7 @@ pub fn init(
},
}),
.request = request,
.outstanding_len = 0,
};
return true;
}
@@ -92,6 +95,8 @@ pub const Opcode = enum(u4) {
binary = 2,
connection_close = 8,
ping = 9,
/// "A Pong frame MAY be sent unsolicited. This serves as a unidirectional
/// heartbeat. A response to an unsolicited Pong frame is not expected."
pong = 10,
_,
};
@@ -106,58 +111,63 @@ pub const ReadSmallTextMessageError = error{
/// Reads the next message from the WebSocket stream, failing if the message does not fit
/// into `recv_buffer`.
pub fn readSmallMessage(ws: *WebSocket) ReadSmallTextMessageError![]u8 {
const header_bytes = (try recv(ws, 2))[0..2];
const h0: Header0 = @bitCast(header_bytes[0]);
const h1: Header1 = @bitCast(header_bytes[1]);
while (true) {
const header_bytes = (try recv(ws, 2))[0..2];
const h0: Header0 = @bitCast(header_bytes[0]);
const h1: Header1 = @bitCast(header_bytes[1]);
 
switch (h0.opcode) {
.text, .binary => {},
.connection_close => return error.ConnectionClose,
else => return error.UnexpectedOpCode,
switch (h0.opcode) {
.text, .binary, .pong => {},
.connection_close => return error.ConnectionClose,
else => return error.UnexpectedOpCode,
}
 
if (!h0.fin) return error.MessageTooBig;
if (!h1.mask) return error.MissingMaskBit;
 
const len: usize = switch (h1.payload_len) {
.len16 => try recvReadInt(ws, u16),
.len64 => std.math.cast(usize, try recvReadInt(ws, u64)) orelse return error.MessageTooBig,
else => @intFromEnum(h1.payload_len),
};
if (len > ws.recv_fifo.buf.len) return error.MessageTooBig;
 
const mask: u32 = @bitCast((try recv(ws, 4))[0..4].*);
const payload = try recv(ws, len);
 
// Skip pongs.
if (h0.opcode == .pong) continue;
 
// The last item may contain a partial word of unused data.
const floored_len = (payload.len / 4) * 4;
const u32_payload: []align(2) u32 = @alignCast(std.mem.bytesAsSlice(u32, payload[0..floored_len]));
for (u32_payload) |*elem| elem.* ^= mask;
const mask_bytes = std.mem.asBytes(&mask)[0 .. payload.len - floored_len];
for (payload[floored_len..], mask_bytes) |*leftover, m| leftover.* ^= m;
 
return payload;
}
 
if (!h0.fin) return error.MessageTooBig;
if (!h1.mask) return error.MissingMaskBit;
 
const len: usize = switch (h1.payload_len) {
.len16 => try recvReadInt(ws, u16),
.len64 => std.math.cast(usize, try recvReadInt(ws, u64)) orelse return error.MessageTooBig,
else => @intFromEnum(h1.payload_len),
};
if (len > ws.recv_fifo.buf.len) return error.MessageTooBig;
 
const mask: u32 = @bitCast((try recv(ws, 4))[0..4].*);
const payload = try recv(ws, len);
 
// The last item may contain a partial word of unused data.
const u32_payload: []align(2) u32 = @alignCast(std.mem.bytesAsSlice(u32, payload));
for (u32_payload) |*elem| elem.* ^= mask;
 
return payload;
}
 
fn recv(ws: *WebSocket, n: usize) ![]u8 {
const result = try recvPeek(ws, n);
ws.recv_fifo.discard(n);
return result;
}
 
const RecvError = std.http.Server.Request.ReadError || error{EndOfStream};
 
fn recvPeek(ws: *WebSocket, n: usize) RecvError![]u8 {
assert(n <= ws.recv_fifo.buf.len);
if (n > ws.recv_fifo.count) {
fn recv(ws: *WebSocket, len: usize) RecvError![]u8 {
ws.recv_fifo.discard(ws.outstanding_len);
assert(len <= ws.recv_fifo.buf.len);
if (len > ws.recv_fifo.count) {
const small_buf = ws.recv_fifo.writableSlice(0);
const needed = n - ws.recv_fifo.count;
const needed = len - ws.recv_fifo.count;
const buf = if (small_buf.len >= needed) small_buf else b: {
ws.recv_fifo.realign();
break :b ws.recv_fifo.writableSlice(0);
};
try @as(RecvError!void, @errorCast(ws.reader.readNoEof(buf)));
ws.recv_fifo.update(buf.len);
const n = try @as(RecvError!usize, @errorCast(ws.reader.readAtLeast(buf, needed)));
if (n < needed) return error.EndOfStream;
ws.recv_fifo.update(n);
}
ws.outstanding_len = len;
// TODO: improve the std lib API so this cast isn't necessary.
return @constCast(ws.recv_fifo.readableSliceOfLen(n));
return @constCast(ws.recv_fifo.readableSliceOfLen(len));
}
 
fn recvReadInt(ws: *WebSocket, comptime I: type) !I {
 
server/main.zig added: 145, removed: 54, total 91
@@ -6,6 +6,7 @@ const Config = @import("Config.zig");
const Allocator = std.mem.Allocator;
const assert = std.debug.assert;
const Db = @import("shared").Db;
const protocol = @import("shared").protocol;
const WebSocket = @import("WebSocket.zig");
const StaticHttpFileServer = @import("StaticHttpFileServer");
 
@@ -190,17 +191,43 @@ pub const Server = struct {
 
while (true) {
const msg = try ws.readSmallMessage();
_ = msg;
@panic("TODO handle message from client");
const cmd = try messageEnum(msg, protocol.Command);
switch (cmd) {
.enqueue_album => {
const album_index = try messageIndex(msg[1..], s.db, Db.Album.Index);
std.log.debug("command: enqueue album {d}", .{album_index});
},
_ => {
std.log.warn("unrecognized command from client: 0x{x}", .{msg[0]});
},
}
}
}
 
fn messageEnum(msg: []const u8, comptime E: type) !E {
comptime assert(@typeInfo(E).Enum.tag_type == u8);
if (msg.len < 1) return error.MessageTruncated;
return @enumFromInt(msg[0]);
}
 
fn messageIndex(msg: []const u8, db: *const Db, comptime E: type) !E {
const I = @typeInfo(E).Enum.tag_type;
const len = switch (E) {
Db.Album.Index => db.albums.entries.len,
else => @compileError("missing type in switch"),
};
if (msg.len < @sizeOf(I)) return error.MessageTruncated;
const int: I = @bitCast(msg[0..@sizeOf(I)].*);
if (int >= len) return error.IndexTooBig;
return @enumFromInt(int);
}
 
fn websocketSendLoop(s: *Server, ws: *WebSocket) !void {
const files = std.mem.sliceAsBytes(s.db.files.items);
const directories = std.mem.sliceAsBytes(s.db.directories.items);
const albums = std.mem.sliceAsBytes(s.db.albums.keys());
const string_bytes = s.db.string_bytes.items;
const Header = @import("shared").protocol.Header;
const Header = protocol.Header;
const header: Header = .{
.files_len = @intCast(s.db.files.items.len),
.directories_len = @intCast(s.db.directories.items.len),
 
shared/protocol.zig added: 145, removed: 54, total 91
@@ -1,6 +1,18 @@
const Db = @import("Db.zig");
 
pub const Header = extern struct {
files_len: u32,
directories_len: u32,
albums_len: u32,
string_bytes_len: u32,
};
 
pub const Command = enum(u8) {
enqueue_album,
_,
};
 
pub const EnqueueAlbum = extern struct {
command: Command = .enqueue_album,
album_index: Db.Album.Index align(1),
};
 
www/main.js added: 145, removed: 54, total 91
@@ -17,14 +17,19 @@
console.log(msg);
},
panic: function (ptr, len) {
const msg = decodeString(ptr, len);
throw new Error("panic: " + msg);
const msg = decodeString(ptr, len);
throw new Error("panic: " + msg);
},
send: function (ptr, len) {
const bytes = new Uint8Array(wasm_exports.memory.buffer, ptr, len)
ws.send(bytes);
},
},
}).then(function(obj) {
wasm_exports = obj.instance.exports;
window.wasm = obj; // for debugging
 
initUi();
connectWebSocket();
});
 
@@ -118,6 +123,7 @@
const title = (dbTitle.length == 0) ? "[Unknown Album]" : dbTitle;
const liDom = domListLibrary.children[i];
liDom.textContent = title;
liDom.attributes['data-id'] = 'a' + i;
}
 
domSectLibrary.classList.remove("hidden");
@@ -131,4 +137,17 @@
listDom.removeChild(listDom.lastChild);
}
}
 
function initUi() {
domListLibrary.addEventListener('click', onClickLibrary, false);
}
 
function onClickLibrary(ev) {
const id = ev.srcElement.attributes['data-id'];
if (id == null) return;
if (id[0] == 'a') {
const albumIndex = +id.substring(1);
wasm_exports.enqueueAlbum(albumIndex);
}
}
})();