/src/Plac.vala at 52424bbaa849773c854f56835731bea534c1db1d

// 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 {
		public class Server : Object {
			/**
			 * IP address of the server.
			 */
			public GLib.InetSocketAddress address { get; construct; }

			/**
			 * TCP port for WebSocket and HTTP connection.
			 */
			public uint16 http_port { get; construct; }

			/**
			 * String uniquely identifies a server among servers in a network.
			 */
			public string id { get; construct; }

			public string version { get; construct; }

			/**
			 * User-facing display name.
			 */
			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 class ServerScanner : Object {
			/**
			 * Found a server.
			 *
			 * ServerScanner emits this signal everytime server returned discovery response:
			 * if a server A returned a response at t1 then returned again at t2, ServerScanner
			 * emits this signal twice (t1 and t2.)
			 */
			public signal void found(Server server);

			/**
			 * Scan has been aborted due to a socket error.
			 */
			public signal void scan_failed(GLib.Error error);

			private GLib.Cancellable? cancellable = null;

			public int64 read_timeout_us = 1000 * 1000 * 2;

			public ServerScanner() {
				Object();
			}

			public void start() {
				GLib.log("Plac", LEVEL_DEBUG, "Starting server scanner...");

				cancellable = new GLib.Cancellable();

				new GLib.Thread<void>("plac-server-scanner", () => {
					try {
						var socket = new GLib.Socket(IPV4, DATAGRAM, UDP);

						socket.set_timeout(3);

						var inet_addr = new GLib.InetAddress.from_bytes(Sood.DISCOVERY_MULTICAST_IPV4_ADDRESS, IPV4);
						var sock_addr = new GLib.InetSocketAddress(inet_addr, Sood.DISCOVERY_SERVER_UDP_PORT);

						while (true) {
							try {
								socket.send_to(sock_addr, (uint8[]) Sood.DISCOVERY_QUERY_PREBUILT, cancellable);

								GLib.log("Plac", LEVEL_DEBUG, "Sent discovery query");
							} catch (GLib.Error error) {
								if (error is GLib.IOError.CANCELLED) {
									return;
								}

								GLib.log("Plac", LEVEL_CRITICAL, "Failed to send discovery query: %s", error.message);

								var e = error.copy();
								GLib.Idle.add(() => {
									scan_failed(e);
									return false;
								});
								cancellable = null;
								return;
							}

							while (true) {
								try {
									GLib.SocketAddress addr;
									var bytes = socket.receive_bytes_from(out addr, 512, read_timeout_us, cancellable);

									if (!(addr is GLib.InetSocketAddress)) {
										GLib.log("Plac", LEVEL_WARNING, "Received discovery response from non-IP socket");
										continue;
									}

									Sood.DiscoveryResponse resp;
									var result = Sood.DiscoveryResponse.parse(out resp, (char[]) bytes.get_data());
									if (result != Sood.Result.OK) {
										GLib.log("Plac", LEVEL_WARNING, "Received malformed discovery response: %s", result.to_string());
										continue;
									}

									// String in Vala is null-terminated, without exception.
									// Using non null-terminated string or casting uint8[] to string
									// will result in out of bound reads. I found no function, class,
									// or language builtins for allocating null-terminated string from
									// non-null one. Slicing is the closest I got.
									var server = new Server(
										(GLib.InetSocketAddress) addr,
										resp.unique_id.slice(0, (long) resp.unique_id_len),
										resp.display_version.slice(0, (long) resp.display_version_len),
										resp.name.slice(0, (long) resp.name_len),
										resp.http_port
									);

									GLib.Idle.add(() => {
										found(server);
										return false;
									});
								} catch (GLib.Error error) {
									if (error is GLib.IOError.TIMED_OUT) {
										break;
									}

									if (error is GLib.IOError.CANCELLED) {
										return;
									}

									GLib.log("Plac", LEVEL_CRITICAL, "Failed to read discovery response: %s", error.message);
									cancellable = null;

									var e = error.copy();
									GLib.Idle.add(() => {
										scan_failed(e);
										return false;
									});
									return;
								}
							}
						}
					} catch (GLib.Error error) {
						GLib.log("Plac", LEVEL_CRITICAL, "Discovery aborted due to an error: %s", error.message);

						var e = error.copy();
						GLib.Idle.add(() => {
							scan_failed(e);
							return false;
						});
					}

					cancellable = null;
				});
			}

			public void stop() {
				GLib.log("Plac", LEVEL_DEBUG, "Stopping server scanner...");
				cancellable.cancel();
			}
		}
	}

	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;
		private bool is_closed = false;
		public size_t retry_max = 3;

		public AsyncConnection(Discovery.Server server) {
			this.conn = new Connection(server, null);
		}

		public AsyncConnection.with_token(Discovery.Server server, string token) {
			this.conn = new Connection(server, token);
		}

		public signal void connection_started();
		public signal void out_of_memory_error();
		public signal void connection_error(ConnectionErrorEvent event);
		public signal void connected(ConnectedEvent event);
		public signal void zones_changed(Transport.ZoneListEvent event);

		public void activate() {
			if (thread != null) {
				return;
			}

			is_closed = false;

			thread = new GLib.Thread<void>("connection-loop", () => {
				GLib.Idle.add(() => {
					connection_started();
					return false;
				});

				size_t retry_count = 0;

				while (true) {
					if (is_closed) {
						return;
					}

					var event = conn.get_event();

					if (event == null) {
						deactivate();
						GLib.Idle.add(() => {
							out_of_memory_error();
							return false;
						});
						return;
					}

					switch (event.kind) {
						case ERROR: {
							var error_event = event.get_connection_error_event();
							if (error_event.code == CLOSED_BY_SERVER && retry_count < retry_max) {
								retry_count += 1;
								GLib.log(
									"Plac", LEVEL_INFO, "Connection closed by server, retrying (#%u)", (uint) retry_count
								);
								break;
							}

							deactivate();
							GLib.Idle.add(() => {
								connection_error(error_event);
								return false;
							});
							break;
						}
						case CONNECTED: {
							retry_count = 0;
							conn.subscribe_zones();
							GLib.Idle.add(() => {
								connected(event.get_connected_event());
								return false;
							});
							break;
						}
						case ZONE_LIST: {
							GLib.Idle.add(() => {
								zones_changed(event.get_zone_list_event());
								return false;
							});
							break;
						}
					}
				}
			});
		}

		public void deactivate() {
			if (thread == null) {
				return;
			}

			// Prevent unnecessary read immediately.
			is_closed = true;

			// Schedule thread disposal. Calling `thread.join` immediately results in
			// `join` from the same thread = deadlock.
			GLib.Idle.add(() => {
				thread.join();
				return false;
			});
		}

		public async void control(Transport.Zone zone, uint16 action) {
			GLib.SourceFunc callback = control.callback;

			new GLib.Thread<void>("control", () => {
				conn.control(zone, action);
				GLib.Idle.add((owned) callback);
			});

			yield;
		}

		public async Transport.SeekResultCode seek(Transport.Zone zone, int64 at_seconds) {
			GLib.SourceFunc callback = seek.callback;
			Transport.SeekResultCode code = UNKNOWN_ERROR;

			new GLib.Thread<void>("seek", () => {
				code = conn.seek(zone, at_seconds);
				GLib.Idle.add((owned) callback);
			});

			yield;
			return code;
		}

		public async Transport.VolumeControlResultCode increase_volume(Transport.Output output) {
			GLib.SourceFunc callback = increase_volume.callback;
			Transport.VolumeControlResultCode code = UNKNOWN_ERROR;

			new GLib.Thread<void>("increase_volume", () => {
				code = conn.increase_volume(output);
				GLib.Idle.add((owned) callback);
			});

			yield;
			return code;
		}

		public async Transport.VolumeControlResultCode decrease_volume(Transport.Output output) {
			GLib.SourceFunc callback = decrease_volume.callback;
			Transport.VolumeControlResultCode code = UNKNOWN_ERROR;

			new GLib.Thread<void>("decrease_volume", () => {
				code = conn.decrease_volume(output);
				GLib.Idle.add((owned) callback);
			});

			yield;
			return code;
		}

		public async Transport.VolumeControlResultCode change_volume(Transport.Output output, double value) {
			GLib.SourceFunc callback = change_volume.callback;
			Transport.VolumeControlResultCode code = UNKNOWN_ERROR;

			new GLib.Thread<void>("change_volume", () => {
				code = conn.change_volume(output, value);
				GLib.Idle.add((owned) callback);
			});

			yield;
			return code;
		}

		public async Browse.Result? browse(Browse.Hierarchy hierarchy, Transport.Zone? zone, Browse.Item? item, bool back) {
			GLib.SourceFunc callback = browse.callback;
			Browse.Result? result = null;

			new GLib.Thread<void>("browse", () => {
				result = conn.browse(hierarchy, zone, item, back);
				GLib.Idle.add((owned) callback);
			});

			yield;
			return (owned) result;
		}

		public string? get_image_url(string image_key, Image.GetOptions options) {
			return conn.get_image_url(image_key, options);
		}
	}
}