Rewrite server resolve process in Vala

Changes

13 changed files (+279/-315)

• src/Moo/Headers.test.vala
  • @@ -27,8 +27,8 @@ var metadata = new Metadata.from_string(input);
    var headers = new Headers.from_string(input, metadata); assert(headers.size == 1); - assert(headers["foo"].size == 1); - assert(headers["foo"][0] == "Bar"); + assert(headers["Foo"].size == 1); + assert(headers["Foo"][0] == "Bar"); assert(headers.last_byte_index == input.length); } catch (Error e) { assert_no_error(e);
  • @@ -47,9 +47,9 @@ var metadata = new Metadata.from_string(input);
    var headers = new Headers.from_string(input, metadata); assert(headers.size == 3); - assert(headers["foo"][0] == "foo-value"); + assert(headers["Foo"][0] == "foo-value"); assert(headers["bar"][0] == "bar-value"); - assert(headers["baz"][0] == "baz-value"); + assert(headers["baZ"][0] == "baz-value"); assert(headers.last_byte_index == input.length); } catch (Error e) { assert_no_error(e);
  • @@ -64,8 +64,8 @@ var metadata = new Metadata.from_string(input);
    var headers = new Headers.from_string(input, metadata); assert(headers.size == 1); - assert(headers["foo"].size == 1); - assert(headers["foo"][0] == "Bar"); + assert(headers["Foo"].size == 1); + assert(headers["Foo"][0] == "Bar"); assert(headers.last_byte_index == input.length); } catch (Error e) { assert_no_error(e);
  • @@ -115,6 +115,25 @@ }
    } catch (Error e) { assert_no_error(e); } + } + }); + + Test.add_func("/libmoo/headers/serialize-back", () => { + try { + var input = "MOO/1 REQUEST foo/bar\nfoo: Bar\nbaz: Qux\n"; + + var metadata = new Metadata.from_string(input); + var headers = new Headers.from_string(input, metadata); + + var serialized = @"$metadata$headers"; + var metadata_after = new Metadata.from_string(serialized); + var headers_after = new Headers.from_string(serialized, metadata_after); + + assert(headers.size == headers_after.size); + assert(headers["foo"][0] == headers_after["foo"][0]); + assert(headers["baz"][0] == headers_after["baz"][0]); + } catch (Error e) { + assert_no_error(e); } }); }
• src/Moo/Headers.vala
  • @@ -32,7 +32,7 @@ public int size { get { return map.size; } }
    public string? content_type { owned get { - var entry = map["content-type"]; + var entry = map["Content-Type"]; if (entry == null) { return null; }
  • @@ -43,7 +43,7 @@ }
    public uint64 content_length { get { - var entry = map["content-length"]; + var entry = map["Content-Length"]; if (entry == null) { return 0; }
  • @@ -59,7 +59,7 @@ }
    public uint64 request_id { get { - var entry = map["request-id"]; + var entry = map["Request-Id"]; if (entry == null) { return 0; }
  • @@ -71,12 +71,18 @@ }
    return uint64.parse(value); } + } + + public Headers() { + Object(); + this.map = new Gee.HashMap<string, Gee.ArrayList<string> >(); } /** * Parses `src`. * - * Header keys will be ASCII lowercased. + * Header keys won't be lowercased or uppercased: Roon server treats + * those header keys in a case sensitive way. * * Throws an `NO_PARSING_CONTEXT` if `meta` is not created by * `Metadata.from_string()`.
  • @@ -105,7 +111,7 @@ if (parts.length < 2) {
    throw new HeadersParseError.NO_DELIMITER("Header delimiter does not found."); } - var key = parts[0].ascii_down().strip(); + var key = parts[0].strip(); if (key.length == 0) { throw new HeadersParseError.EMPTY_KEY("Found an empty header key."); }
  • @@ -134,6 +140,31 @@ }
    public new Gee.ArrayList<string>? @get(string key) { return map[key]; + } + + public string to_string() { + var builder = new GLib.StringBuilder(); + + foreach (var entry in map) { + foreach (var value in entry.value) { + builder.append_printf("%s: %s\n", entry.key, value); + } + } + + builder.append("\n"); + + return builder.str; + } + + public void write(string key, string value) { + var existing = map[key]; + if (existing != null) { + existing.add(value); + } else { + var entry = new Gee.ArrayList<string>(); + entry.add(value); + map[key] = entry; + } } } }
• src/Moo/Metadata.test.vala
  • @@ -193,6 +193,16 @@ }
    } } }); + + Test.add_func("/libmoo/metadata/seriealize-back", () => { + try { + var input = "MOO/1 REQUEST foo/bar\n"; + var metadata = new Metadata.from_string(input); + assert(input == metadata.to_string()); + } catch (Error e) { + assert_no_error(e); + } + }); } } }
• src/Moo/Metadata.vala
  • @@ -30,8 +30,8 @@ public string service { get; construct; }
    public int last_byte_index { get; construct; default = -1; } - public Metadata(uint32 version = 1, string verb, string service) { - Object(version: version, verb: verb, service: service); + public Metadata(string verb, string service) { + Object(version: 1, verb: verb, service: service); } public Metadata.from_string(string src) throws MetadataParseError {
  • @@ -91,6 +91,10 @@
    i = next_lf + 1; Object(version: (uint32) version, verb: verb, service: service, last_byte_index: i); + } + + public string to_string() { + return "%s%u %s %s\n".printf(SIGNATURE, version, verb, service); } } }
• src/Plac.vala
  • @@ -15,21 +15,6 @@ //
    // SPDX-License-Identifier: Apache-2.0 namespace Plac { - namespace Discovery { - public async Plac.Discovery.ScanResult? resolve_async(string server_id, string ip_addr, uint16 http_port) { - GLib.SourceFunc callback = resolve_async.callback; - Plac.Discovery.ScanResult? result = null; - - new GLib.Thread<void>("server-resolve", () => { - result = Plac.Discovery.resolve(server_id, ip_addr, http_port); - GLib.Idle.add((owned) callback); - }); - - yield; - return (owned) result; - } - } - public class AsyncConnection : GLib.Object { private Connection conn; private GLib.Thread<void>? thread = null;
• src/Plac/Server.vala
  • @@ -16,6 +16,12 @@ // SPDX-License-Identifier: Apache-2.0
    namespace Plac { namespace V2 { + public errordomain ResolveServerError { + SERVER_NOT_FOUND, + SERVER_MISMATCH, + NETWORK_ERROR, + } + public class Server : Object { /** * IP address of the server.
  • @@ -41,6 +47,93 @@ public string name { get; construct; }
    public Server(GLib.InetSocketAddress address, string id, string version, string name, uint16 http_port) { Object(address: address, id: id, version: version, name: name, http_port: http_port); + } + + public static async Server from_address( + GLib.InetSocketAddress address, uint16 http_port, string id, GLib.Cancellable? cancellable = null + ) throws ResolveServerError { + var session_singleton = new Session(); + var session = session_singleton.session; + + var url = GLib.Uri.build( + NONE, "ws:", null, address.address.to_string(), http_port, + "/api", null, null + ); + var msg = new Soup.Message.from_uri("GET", url); + + Soup.WebsocketConnection conn; + try { + // Roon API does not specify WebSocket subprotocols. + conn = yield session.websocket_connect_async( + msg, null, null, GLib.Priority.DEFAULT, cancellable + ); + } catch (GLib.Error error) { + GLib.log("Plac", LEVEL_DEBUG, "Handshake error"); + throw new ResolveServerError.NETWORK_ERROR(error.message); + } + + var message = yield get_message(conn); + + Moo.Metadata meta; + Moo.Headers headers; + Roon.Registry.Info.Response info; + try { + meta = new Moo.Metadata.from_string(message); + headers = new Moo.Headers.from_string(message, meta); + if (headers.request_id != 1) { + throw new ResolveServerError.SERVER_MISMATCH("Request-Id does not match."); + } + if (headers.content_type != "application/json") { + throw new ResolveServerError.SERVER_MISMATCH("Not a JSON body."); + } + var body = new Moo.JsonBody.from_string(message, headers); + info = new Roon.Registry.Info.Response.from_json(body.data); + } catch (GLib.Error error) { + throw new ResolveServerError.SERVER_MISMATCH(error.message); + } + + if (info == null) { + throw new ResolveServerError.SERVER_MISMATCH("Got invalid info response"); + } + + if (info.core_id != id) { + throw new ResolveServerError.SERVER_MISMATCH("Server at the location has different server ID."); + } + + return new Server(address, id, info.display_version, info.display_name, http_port); + } + + private static async string get_message(Soup.WebsocketConnection conn) throws ResolveServerError { + SourceFunc callback = get_message.callback; + GLib.Bytes? bytes = null; + + conn.message.connect((c, type, b) => { + bytes = b; + callback(); + }); + + conn.error.connect((c, error) => { + callback(); + }); + + conn.send_binary(build_request().data); + + yield; + + if (bytes == null) { + throw new ResolveServerError.SERVER_NOT_FOUND("Server not found."); + } + + return (string) GLib.Bytes.unref_to_data(bytes); + } + + private static string build_request() { + var meta = new Moo.Metadata("REQUEST", "com.roonlabs.registry:1/info"); + + var headers = new Moo.Headers(); + headers.write("Request-Id", "1"); + + return @"$meta$headers"; } } }
• src/Plac/Session.vala (new)
  • @@ -0,0 +1,30 @@
    + // Copyright 2025 Shota FUJI + // + // Licensed under the Apache License, Version 2.0 (the "License"); + // you may not use this file except in compliance with the License. + // You may obtain a copy of the License at + // + // http://www.apache.org/licenses/LICENSE-2.0 + // + // Unless required by applicable law or agreed to in writing, software + // distributed under the License is distributed on an "AS IS" BASIS, + // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + // See the License for the specific language governing permissions and + // limitations under the License. + // + // SPDX-License-Identifier: Apache-2.0 + + namespace Plac { + namespace V2 { + [SingleInstance] + private class Session : Object { + public Soup.Session session = new Soup.Session.with_options( + "user_agent", "plac-for-gtk4/0.0 " + ); + + public Session() { + Object(); + } + } + } + }
• src/Roon/api.zig
  • @@ -15,6 +15,7 @@ //
    // SPDX-License-Identifier: Apache-2.0 const browse = @import("./services/browse.zig"); + const registry = @import("./services/registry.zig"); pub const public_api: []const type = &.{ browse.Hierarchy,
  • @@ -28,4 +29,5 @@ browse.browse.Request,
    browse.browse.Response, browse.load.Request, browse.load.Response, + registry.info.Response, };
• src/Roon/services/registry.zig (new)
  • @@ -0,0 +1,62 @@
    + // Copyright 2025 Shota FUJI + // + // Licensed under the Apache License, Version 2.0 (the "License"); + // you may not use this file except in compliance with the License. + // You may obtain a copy of the License at + // + // http://www.apache.org/licenses/LICENSE-2.0 + // + // Unless required by applicable law or agreed to in writing, software + // distributed under the License is distributed on an "AS IS" BASIS, + // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + // See the License for the specific language governing permissions and + // limitations under the License. + // + // SPDX-License-Identifier: Apache-2.0 + + const std = @import("std"); + + const rc = @import("../rc.zig"); + const json = @import("../json.zig"); + + const allocator = std.heap.c_allocator; + + pub const info = struct { + pub const Response = json.Deserializable(rc.RefCounted(extern struct { + pub const cname = "roon_registry_info_Response"; + pub const deserializable = true; + + core_id: [*:0]const u8, + display_name: [*:0]const u8, + display_version: [*:0]const u8, + + pub fn init(raw: Raw) !@This() { + const core_id = try allocator.dupeZ(u8, raw.core_id); + errdefer allocator.free(core_id); + + const display_name = try allocator.dupeZ(u8, raw.display_name); + errdefer allocator.free(display_name); + + const display_version = try allocator.dupeZ(u8, raw.display_version); + errdefer allocator.free(display_version); + + return .{ + .core_id = core_id.ptr, + .display_name = display_name.ptr, + .display_version = display_version.ptr, + }; + } + + pub const Raw = struct { + core_id: []const u8, + display_name: []const u8, + display_version: []const u8, + }; + + pub fn deinit(self: *const @This()) void { + allocator.free(std.mem.span(self.display_version)); + allocator.free(std.mem.span(self.display_name)); + allocator.free(std.mem.span(self.core_id)); + } + }), .{}); + };
• src/Windows/MainWindow.vala
  • @@ -150,10 +150,22 @@ listen_events();
    return; } - resolve_server_with_addr.begin(ip_addr, port, (obj, res) => { + var colon = ip_addr.index_of(":"); + var patched_ip_addr = colon < 0 ? ip_addr : ip_addr.slice(0, colon); + var address = new GLib.InetSocketAddress.from_string(patched_ip_addr, port); + Plac.V2.Server.from_address.begin(address, port, server_id, null, (obj, res) => { try { - resolve_server_with_addr.end(res); - } catch (ResolveError e) { + var server = Plac.V2.Server.from_address.end(res); + this.server = new Plac.Discovery.Server( + server.id, + server.name, + server.version, + server.address.to_string(), + server.http_port + ); + conn = new Plac.AsyncConnection.with_token(this.server, settings.connected_server_token); + } catch (Plac.V2.ResolveServerError error) { + GLib.log("Plac", LEVEL_DEBUG, "Failed to resolve: %s", error.message); GLib.log("Plac", LEVEL_INFO, "Failed to restore connection, scanning server"); try_listen(); return;
  • @@ -228,41 +240,6 @@ });
    yield; scanner.stop(); - - if (error != null) { - throw error; - } - } - - private async void resolve_server_with_addr(string ip_addr, uint16 http_port) throws ResolveError { - GLib.SourceFunc callback = resolve_server_with_addr.callback; - ResolveError? error = null; - - Plac.Discovery.resolve_async.begin(server_id, ip_addr, http_port, (obj, res) => { - var result = Plac.Discovery.resolve_async.end(res); - - if (result.code != OK) { - GLib.log("Plac", LEVEL_INFO, "Failed to connect to server at %s:%u: %s", - ip_addr, http_port, result.code.to_string()); - Idle.add((owned) callback); - error = new ResolveError.CONNECTION_ERROR("Failed to connect: %s".printf(result.code.to_string())); - return; - } - - if (result.servers.length < 1) { - GLib.log("Plac", LEVEL_CRITICAL, "Server not found at %s:%u: ID=%s", - ip_addr, http_port, server_id); - Idle.add((owned) callback); - error = new ResolveError.SERVER_NOT_FOUND("Server not found"); - return; - } - - server = result.servers[0]; - conn = new Plac.AsyncConnection.with_token(server, settings.connected_server_token); - Idle.add((owned) callback); - }); - - yield; if (error != null) { throw error;
• src/core/discovery.zig
  • @@ -182,188 +182,6 @@ @export(&new, .{ .name = std.fmt.comptimePrint("{s}_new", .{cname}) });
    } }; - pub const ScanResult = extern struct { - const cname = "plac_discovery_scan_result"; - const allocator = std.heap.c_allocator; - - internal: *Internal, - - servers_ptr: [*]*Server, - servers_len: usize, - code: Code = .unknown, - - pub const Code = enum(c_int) { - ok = 0, - unknown = 1, - network_unavailable = 2, - socket_permission_denied = 3, - socket_error = 4, - out_of_memory = 5, - }; - - pub const Internal = struct { - arc: Arc = .{}, - }; - - pub fn make() !*ScanResult { - const self = try allocator.create(ScanResult); - errdefer allocator.destroy(self); - - const internal = try allocator.create(Internal); - errdefer allocator.destroy(internal); - - const servers: []*Server = &.{}; - - internal.* = .{}; - - self.* = .{ - .internal = internal, - .servers_ptr = servers.ptr, - .servers_len = servers.len, - }; - - return self; - } - - pub fn initFindResult(self: *ScanResult, found: *Server) std.mem.Allocator.Error!void { - const servers = try allocator.alloc(*Server, 1); - errdefer allocator.free(servers); - - servers[0] = found.retain(); - - self.servers_ptr = servers.ptr; - self.servers_len = 1; - self.code = .ok; - } - - pub fn retain(ptr: ?*ScanResult) callconv(.C) *ScanResult { - var self = ptr orelse @panic(std.fmt.comptimePrint( - "Received null pointer on {s}_{s}", - .{ cname, @src().fn_name }, - )); - self.internal.arc.ref(); - return self; - } - - pub fn release(ptr: ?*ScanResult) callconv(.C) void { - var self = ptr orelse @panic(std.fmt.comptimePrint( - "Received null pointer on {s}_{s}", - .{ cname, @src().fn_name }, - )); - - if (self.internal.arc.unref()) { - freelog(self); - - const servers = self.servers_ptr[0..self.servers_len]; - for (servers) |server| { - server.release(); - } - allocator.free(servers); - allocator.destroy(self.internal); - allocator.destroy(self); - } - } - - pub fn export_capi() void { - @export(&retain, .{ .name = std.fmt.comptimePrint("{s}_retain", .{cname}) }); - @export(&release, .{ .name = std.fmt.comptimePrint("{s}_release", .{cname}) }); - } - }; - - pub fn resolve(server_id: [*:0]const u8, ip_addr: [*:0]const u8, http_port: u16) callconv(.C) ?*ScanResult { - var arena = std.heap.ArenaAllocator.init(std.heap.c_allocator); - defer arena.deinit(); - - const result = ScanResult.make() catch { - return null; - }; - - const id = std.mem.span(server_id); - const addr = std.mem.span(ip_addr); - - resolveInternal(arena.allocator(), id, addr, http_port, result) catch |err| { - result.code = switch (err) { - error.OutOfMemory => .out_of_memory, - else => .unknown, - }; - }; - - return result.retain(); - } - - fn resolveInternal( - allocator: std.mem.Allocator, - server_id: []const u8, - ip_addr: []const u8, - http_port: u16, - result: *ScanResult, - ) !void { - std.log.debug("Establishing WebSocket connection to {s}:{d}...", .{ - ip_addr, http_port, - }); - - var ws = try websocket.Client.init(allocator, .{ - .host = ip_addr, - .port = http_port, - .max_size = std.math.maxInt(u32), - }); - defer ws.deinit(); - - std.log.debug("Performing WebSocket handshake...", .{}); - ws.handshake("/api", .{ .timeout_ms = 1_000 }) catch |err| { - std.log.warn("Failed to perform WebSocket handshake, fallback to scan: {s}", .{@errorName(err)}); - result.code = .ok; - return; - }; - - std.log.debug("Requesting server info...", .{}); - - { - const req = try RegistryService.Info.request(allocator, 0); - defer allocator.free(req); - try ws.writeBin(req); - } - - const msg = (try ws.read()) orelse { - return error.ConnectionClosed; - }; - defer ws.done(msg); - - _, const header_ctx = try moo.Metadata.parse(msg.data); - const resp = try RegistryService.Info.response(allocator, header_ctx, msg.data); - defer resp.deinit(); - - if (!std.mem.eql(u8, server_id, resp.value.core_id)) { - std.log.info( - "Found Roon Server on saved IP address, got different server instance, fallback to scan: saved ID={s}, found ID={s}", - .{ - server_id, - resp.value.core_id, - }, - ); - result.code = .ok; - return; - } - - const addr = try std.net.Address.parseIp4(ip_addr, http_port); - - const sood_resp = sood.discovery.Response{ - .display_version = resp.value.display_version, - .name = resp.value.display_name, - .unique_id = server_id, - .http_port = http_port, - }; - - var server = try Server.make(&sood_resp, &addr); - errdefer server.release(); - - try result.initFindResult(server); - return; - } - pub fn export_capi() void { Server.export_capi(); - ScanResult.export_capi(); - - @export(&resolve, .{ .name = "plac_discovery_resolve" }); }
• src/core/plac.h
  • @@ -48,33 +48,6 @@ const char *ip_addr,
    uint16_t http_port ); - // discovery.ScanResult.Code - typedef enum { - PLAC_DISCOVERY_SCAN_RESULT_OK = 0, - PLAC_DISCOVERY_SCAN_RESULT_UNKNOWN = 1, - PLAC_DISCOVERY_SCAN_RESULT_NETWORK_UNAVAILABLE = 2, - PLAC_DISCOVERY_SCAN_RESULT_SOCKET_PERMISSION_DENIED = 3, - PLAC_DISCOVERY_SCAN_RESULT_SOCKET_ERROR = 4, - PLAC_DISCOVERY_SCAN_RESULT_OUT_OF_MEMORY = 5, - } plac_discovery_scan_result_code; - - // discovery.ScanResult - typedef struct { - void *__pri; - plac_discovery_server **servers_ptr; - size_t servers_len; - plac_discovery_scan_result_code code; - } plac_discovery_scan_result; - plac_discovery_scan_result *plac_discovery_scan_result_retain(plac_discovery_scan_result*); - void plac_discovery_scan_result_release(plac_discovery_scan_result*); - - // discovery - plac_discovery_scan_result *plac_discovery_resolve( - const char *server_id, - const char *ip_addr, - uint16_t http_port - ); - // transport.NowPlaying typedef struct { void *__pri;
• src/core/plac.vapi
  • @@ -38,46 +38,6 @@ public string version;
    public string ip_addr; public uint16 http_port; } - - [CCode ( - cname = "plac_discovery_scan_result_code", - cprefix = "PLAC_DISCOVERY_SCAN_RESULT_", - has_type_id = false - )] - public enum ScanResultCode { - OK = 0, - UNKNOWN = 1, - NETWORK_UNAVAILABLE = 2, - SOCKET_PERMISSION_DENIED = 3, - SOCKET_ERROR = 4, - OUT_OF_MEMORY = 5, - } - - [CCode ( - cname = "plac_discovery_scan_result", - ref_function = "plac_discovery_scan_result_retain", - unref_function = "plac_discovery_scan_result_release" - )] - [Compact] - public class ScanResult { - [CCode (cname = "plac_discovery_scan_result_retain")] - public void @ref (); - - [CCode (cname = "plac_discovery_scan_result_release")] - public void unref (); - - [CCode ( - cname = "servers_ptr", - array_length_cname = "servers_len", - array_length_type = "size_t" - )] - public Server[] servers; - - public ScanResultCode code; - } - - [CCode (cname = "plac_discovery_resolve")] - public ScanResult? resolve(string server_id, string ip_addr, uint16 http_port); } namespace Transport {