commit d1a98c5da2d1f1eb6ef2499b354041cf7ac0e5a1 Author: Matteo Rosati Date: Tue Aug 11 10:55:02 2026 +0200 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b56bb8c --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +syncplay diff --git a/CLARIFY.md b/CLARIFY.md new file mode 100644 index 0000000..18468a1 --- /dev/null +++ b/CLARIFY.md @@ -0,0 +1,134 @@ +## Missing specifications that block implementation + +1. Media selection + +The client command has no movie argument. It is unclear whether the user: + +- passes a file to syncplay; +- opens a file afterward through mpv; +- or launches mpv separately. + +ANSWER: passes a file to syncplay + +2. Session model + +The server broadcasts to “all connected clients,” but it does not say whether there is exactly one global viewing session or whether named rooms are required. + +ANSWER: no rooms. one global viewing session. This program is intended to be run by people that know each other, and maximum 2-3 persons. + +3. Joining behavior + +A newly connected client needs an initial state: current position, paused/playing status, and possibly media identity. A pure event dispatcher cannot provide that unless another client republishes its state. + +ANSWER: `syncplay` starts and has a movie specified. The initial state is paused and at 00:00:00. + +4. Authoritative state + +There is no rule for simultaneous or conflicting actions. For example, A pauses while B seeks or immediately presses play. The server needs an ordering rule, even if it is only “last event received wins.” + +ANSWER: even if the movie is paused, users can still seek. This is how normally the video player works. So there is no reason why a user could seek while the player is paused. Agree? + +5. Network protocol + +The plan does not specify: + +- TCP, WebSocket, or another transport; +- message framing; +- message schema; +- protocol versioning; +- connection handshake; - keepalive or timeout behavior. + +ANSWER: TCP preferrable, but pick the easiest option. For all other above points i have no answer, pick sensible defaults. + +3. Feedback-loop prevention + +Remotely applying pause or seek causes mpv events locally. The plan does not define how clients recognize and suppress those echoed changes. + +ANSWER: i don't understand where is the issue. The event is firts applied locally (i.e. pause). Then that same event is sent to the server for the other clients to pick it up. + +4. Meaning of synchronization + +Relaying play, pause, and seek operations does not maintain close synchronization during uninterrupted playback. Independent players can gradually drift, and network latency makes “play now” happen at different times. +The plan must say whether the prototype only mirrors user actions or also periodically measures and corrects drift. + +ANSWER: I don't care about the exact synchronization or drifting. If drifting happens, the users can still seek and all other clients will receive the event. + +5. Seek semantics + +“Seek to xx:xx:xx” suggests absolute seconds, but the protocol needs to specify: + +- absolute versus relative positions; +- precision; +- exact versus keyframe seek; +- behavior when duration differs between clients. + +ANSWER: OK. + +6. Source compatibility + +There is no requirement that participants load the same media. A position alone is meaningless if files, editions, durations, or opening offsets differ. + +ANSWER: it is assumed that the clients pick the same file. + +7. Reconnect behavior + +It is unspecified whether a disconnected client retries, exits, keeps playing locally, or resynchronizes after reconnecting. + +ANSWER: if a client disconnects, or loses the connection to the server, the playback continues normally. Log the event to the console. + +8. mpv startup lifecycle + +The plan does not define: + +- how long to wait for the IPC socket; +- what happens if mpv fails to start; +- how stale socket files are handled; +- whether closing mpv terminates syncplay; +- whether closing syncplay terminates mpv. + +ANSWERS: + +- how long to wait for the IPC socket: I don't understand the question. Use sensible defaults. +- what happens if mpv fails to start: Quit and log the error. +- how stale socket files are handled: Quit and log the error. +- whether closing mpv terminates syncplay: Yes. +- whether closing syncplay terminates mpv: Yes. + +9. Platform scope + +/Applications/mpv.app/... is macOS-specific, while Unix sockets suggest macOS/Linux. Supported client platforms and executable discovery need to be explicit. + +ANSWER: Support only MacOS and Linux (Linux only for server) with the defaults i have provided. + +10. Configuration + +Hard-coding /tmp/mpv-socket prevents multiple local instances and risks collision with a stale or unrelated socket. The mpv executable and socket should at least be overrideable, even if defaults are provided. + +ANSWER: I don't care about multiple local instances. Only one is expected. Provide `--socket` and `--mpv` cli options to override defaults. + +11. Security boundary + +A server bound to 0.0.0.0 permits anyone who can reach it to control every connected player unless authentication or network-level trust is assumed. The plan needs to state whether an unauthenticated trusted-network prototype is acceptable. + +ANSWER: I don't care about security or authentication. This is expected to run on a trusted environment by a few people. + +12. Failure handling and limits + +Missing behavior includes malformed messages, slow clients, disconnected sockets, oversized messages, and server capacity. + +ANSWER: TCP or WebSocket should deal with malformed messages already. IDC about slow clients, we are talking about very small packets. Oversized messages and server capacity are not a concern for all answers above. + +13. Prototype acceptance criteria + +There is no concrete test defining “up and running”: supported OS, number of clients, acceptable latency/drift, reconnect expectations, and required automated tests. + +ANSWER: supported OS: MacOS and Linux (Linux server only). I said already IDC about drifting. Write automated unit tests in Odin where applicable. I don't need end-to-end testing. + +## Internal inconsistencies or misleading details + +- The IPC socket path is supplied to mpv, but mpv creates/listens on it; syncplay does not create a shared read/write command stream. ANSWER IDK, pick defaults. +- “Every interaction with MPV will write in the socket” is incorrect; explicit property observation and event handling are needed. ANSWER IDK, pick defaults. +- The example contains --input-terminal=no twice. ANSWER: remove the duplicate option, it is a typo. +- “The socket path and mpv path must likely be constants” conflicts with portability and safe concurrent use. ANSWER: we provide overrides via cli options. +- Calling the server only an “events dispatcher” conflicts with seamless late joining, which requires either server-held state or a designated client supplying a snapshot. ANSWER: IDC about late joining. If someone joins late, a seek event from any client will fix the drift. +- “Synchronized playback” implies clock/drift synchronization, while the listed behavior only promises event replication. ANSWER IDC about clock/drift sync. diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..8d4e0c3 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,83 @@ +# Sync play for MPV + +Goal: write an Odin application to sync playback between remote viewers via a small Odin program. + +The program will be called `syncplay`. The player is called `mpv`. `syncplay` is a single executable that can be run by clients, and is also executable in a remote server that functions as events dispatcher. The server is required. + +Use case: two friends want to see the same movie using MPV, and they want synchronized playback. If one pauses, the video pauses for the other viewer. If one seeks to xx:xx:xx, also the other player seeks to xx:xx:xx. + +**Server** + +```sh +./syncplay server --port --host 0.0.0.0 +``` + +**Client** + +```sh +./syncplay client --port --host \ + [--mpv ] [--socket ] +``` + +Supported operations: + +- play +- pause +- seek + +On macOS, the client launches MPV with a JSON IPC UNIX socket. `syncplay` +connects to that socket, observes pause changes and seek events, and sends JSON +commands to apply events received from the relay server. + +The TCP protocol is newline-delimited JSON. There is one global session, no +authentication, and no server-held playback state. Messages are absolute state +changes: + +```json +{"version":1,"type":"welcome","client_id":1} +{"version":1,"type":"pause","paused":true} +{"version":1,"type":"pause","paused":false} +{"version":1,"type":"seek","position":123.456} +``` + +The server assigns a numeric ID when a client connects and includes the +triggering ID in every relayed event. Both server and client terminals log each +play, pause, and seek action with that client ID. + +Clients start paused at `00:00:00` and are assumed to use the same movie. There +is no clock synchronization or drift correction. If a connection is lost, +playback continues locally and the event is logged; clients do not reconnect. + +**Example flow**: + +0. Someone starts `syncplay` on a remote server. +1. User A starts `syncplay`. The program starts and runs the `mpv` player. +2. Then it connects to the remote `syncplay` instance. +3. `syncplay` observes pause and seek events reported by MPV over JSON IPC. +4. `syncplay` translates supported local events and sends them to the server. +5. The server will stream to all connected clients the same event. +6. Any connected client receives the event and sends the corresponding JSON IPC + command to its local MPV instance. + +Remote seeks use a serialized, acknowledged transaction: disable delivery of +the `seek` event for the IPC connection, apply an `absolute+exact` seek, then +re-enable the event even when the seek fails. This prevents remote seeks from +being reported as new local seeks and looping through the server. + +**Example cli command to run MPV** + +Pay attention to the socket path. That, along with the MPV path must likely be constants in the Odin program. + +```sh +/Applications/mpv.app/Contents/MacOS/mpv \ + --force-window=yes \ + --idle=yes \ + --pause=yes \ + --start=0 \ + --hr-seek=always \ + --keep-open=always \ + --keep-open-pause=yes \ + --input-ipc-server=/tmp/mpv-socket \ + --input-terminal=no \ + --terminal=no +``` diff --git a/README.md b/README.md new file mode 100644 index 0000000..56e84b8 --- /dev/null +++ b/README.md @@ -0,0 +1,88 @@ +# syncplay + +`syncplay` is a small Odin program that relays play, pause, and seek actions +between a few trusted mpv users. One executable provides both the TCP relay +server and the macOS client. + +## Requirements + +- Odin 2026-07 or newer +- macOS for client mode +- macOS or Linux for server mode +- mpv installed on every client +- The same media file (including the same cut) on every client + +## Build and test + +```sh +odin check . +odin test . +odin build . -out:syncplay +``` + +## Run + +Start one relay server: + +```sh +./syncplay server --host 0.0.0.0 --port 9000 +``` + +Then each viewer starts a client with their local copy of the movie: + +```sh +./syncplay client \ + --host server.example.test \ + --port 9000 \ + /path/to/movie.mkv +``` + +The macOS defaults can be overridden: + +```sh +./syncplay client \ + --host server.example.test \ + --port 9000 \ + --mpv /custom/path/to/mpv \ + --socket /tmp/custom-mpv-socket \ + /path/to/movie.mkv +``` + +The client starts at `00:00:00` in the paused state. Closing mpv closes the +client. Interrupting the client terminates the mpv process it launched. If the +server connection is lost after startup, mpv continues normally and the client +logs that synchronization has stopped. + +## Protocol + +The trusted-network protocol is newline-delimited JSON over TCP: + +```json +{"version":1,"type":"welcome","client_id":1} +{"version":1,"type":"pause","paused":true} +{"version":1,"type":"pause","paused":false} +{"version":1,"type":"seek","position":3723.125} +``` + +Messages are limited to 4096 bytes. The server validates and re-encodes every +message, preserves processing order, and broadcasts it to every connection +except its sender. It assigns each connection a numeric client ID and adds the +triggering ID to relayed events. Server and client terminals log play, pause, +and seek actions with that ID; the triggering client is marked with `(you)` in +its own terminal. + +Remote seeks are applied using an acknowledged mpv IPC transaction: seek event +delivery is disabled for this IPC connection, an exact absolute seek is sent, +and event delivery is restored even if the seek fails. This prevents a remote +seek from being mistaken for a new local seek and sent back indefinitely. + +## Prototype limitations + +- One global session; no rooms or late-join state snapshot +- No authentication, encryption, or Internet-facing security +- No automatic reconnection +- No media identity checking +- No clock synchronization or drift correction +- mpv can report internally generated operations as seek events +- A local seek made during the brief remote-seek suppression transaction may + not be relayed because mpv events do not identify their origin diff --git a/cli.odin b/cli.odin new file mode 100644 index 0000000..86f14ab --- /dev/null +++ b/cli.odin @@ -0,0 +1,122 @@ +package main + +import "core:fmt" +import "core:os" +import "core:strconv" +import "core:strings" + +DEFAULT_SERVER_HOST :: "0.0.0.0" +DEFAULT_MPV_PATH :: "/Applications/mpv.app/Contents/MacOS/mpv" +DEFAULT_SOCKET_PATH :: "/tmp/mpv-socket" + +Command_Kind :: enum { + None, + Server, + Client, + Help, +} + +Options :: struct { + command: Command_Kind, + host: string, + port: int, + mpv_path: string, + socket_path: string, + movie_path: string, +} + +usage :: proc() { + fmt.println("syncplay - synchronize basic mpv playback events") + fmt.println("") + fmt.println("Usage:") + fmt.println(" syncplay server --port [--host 0.0.0.0]") + fmt.println(" syncplay client --host --port [--mpv ] [--socket ] ") +} + +parse_cli :: proc(args: []string) -> (options: Options, error_message: string) { + options.mpv_path = DEFAULT_MPV_PATH + options.socket_path = DEFAULT_SOCKET_PATH + + if len(args) < 2 { + return options, "expected server or client subcommand" + } + switch args[1] { + case "server": + options.command = .Server + options.host = DEFAULT_SERVER_HOST + case "client": + options.command = .Client + case "help", "--help", "-h": + options.command = .Help + return options, "" + case: + return options, fmt.tprintf("unknown subcommand %q", args[1]) + } + + i := 2 + for i < len(args) { + arg := args[i] + if strings.has_prefix(arg, "--") { + if i+1 >= len(args) { + return options, fmt.tprintf("missing value for %s", arg) + } + value := args[i+1] + switch arg { + case "--host": + options.host = value + case "--port": + port, ok := strconv.parse_int(value) + if !ok || port < 1 || port > 65535 { + return options, "port must be an integer from 1 to 65535" + } + options.port = port + case "--mpv": + if options.command != .Client { + return options, "--mpv is only valid for client" + } + options.mpv_path = value + case "--socket": + if options.command != .Client { + return options, "--socket is only valid for client" + } + options.socket_path = value + case: + return options, fmt.tprintf("unknown option %s", arg) + } + i += 2 + continue + } + + if options.command != .Client || options.movie_path != "" { + return options, fmt.tprintf("unexpected positional argument %q", arg) + } + options.movie_path = arg + i += 1 + } + + if options.port == 0 { + return options, "--port is required" + } + if options.command == .Client { + if options.host == "" { + return options, "--host is required for client" + } + if options.movie_path == "" { + return options, "movie path is required" + } + } + return options, "" +} + +validate_client_paths :: proc(options: Options) -> (error_message: string) { + if _, err := os.stat(options.movie_path, context.temp_allocator); err != nil { + return fmt.tprintf("movie does not exist: %s", options.movie_path) + } + if _, err := os.stat(options.mpv_path, context.temp_allocator); err != nil { + return fmt.tprintf("mpv does not exist: %s", options.mpv_path) + } + if _, err := os.stat(options.socket_path, context.temp_allocator); err == nil { + return fmt.tprintf("socket path already exists: %s", options.socket_path) + } + return "" +} diff --git a/cli_test.odin b/cli_test.odin new file mode 100644 index 0000000..4972e3f --- /dev/null +++ b/cli_test.odin @@ -0,0 +1,30 @@ +package main + +import "core:testing" + +@(test) +cli_parses_server :: proc(t: ^testing.T) { + args := []string{"syncplay", "server", "--port", "9000"} + options, err := parse_cli(args) + testing.expect_value(t, err, "") + testing.expect_value(t, options.command, Command_Kind.Server) + testing.expect_value(t, options.host, DEFAULT_SERVER_HOST) + testing.expect_value(t, options.port, 9000) +} + +@(test) +cli_parses_client :: proc(t: ^testing.T) { + args := []string{"syncplay", "client", "--host", "example.test", "--port", "9000", "movie.mkv"} + options, err := parse_cli(args) + testing.expect_value(t, err, "") + testing.expect_value(t, options.command, Command_Kind.Client) + testing.expect_value(t, options.host, "example.test") + testing.expect_value(t, options.movie_path, "movie.mkv") +} + +@(test) +cli_rejects_bad_port :: proc(t: ^testing.T) { + args := []string{"syncplay", "server", "--port", "70000"} + _, err := parse_cli(args) + testing.expect(t, err != "") +} diff --git a/client.odin b/client.odin new file mode 100644 index 0000000..ab2df80 --- /dev/null +++ b/client.odin @@ -0,0 +1,448 @@ +package main + +import "base:runtime" +import "core:encoding/json" +import "core:fmt" +import "core:net" +import "core:os" +import "core:sync" +import "core:sys/posix" +import "core:thread" +import "core:time" + +Client_State :: struct { + mutex: sync.Mutex, + client_id: u64, + + network_socket: net.TCP_Socket, + network_active: bool, + network_write_mutex: sync.Mutex, + + mpv: ^Mpv_Connection, + process: os.Process, + stopping: bool, + + ignore_initial_pause: bool, + remote_pause_pending: bool, + remote_pause_value: bool, + have_last_pause: bool, + last_pause: bool, + local_seek_pending: bool, +} + +client_close_network :: proc(client: ^Client_State, reason: string) { + sync.mutex_lock(&client.mutex) + if !client.network_active { + sync.mutex_unlock(&client.mutex) + return + } + client.network_active = false + socket := client.network_socket + sync.mutex_unlock(&client.mutex) + + _ = net.shutdown(socket, .Both) + net.close(socket) + fmt.printfln("client: server connection lost (%s); playback will continue", reason) +} + +client_send_event :: proc(client: ^Client_State, event: Playback_Event) -> bool { + line := encode_protocol_event(event) + defer delete(line) + + sync.mutex_lock(&client.network_write_mutex) + defer sync.mutex_unlock(&client.network_write_mutex) + sync.mutex_lock(&client.mutex) + active := client.network_active + socket := client.network_socket + client_id := client.client_id + sync.mutex_unlock(&client.mutex) + log_playback_event("client", client_id, event, is_local = true) + if !active { + return false + } + + written, send_err := net.send_tcp(socket, transmute([]byte)line) + if send_err != nil || written != len(line) { + client_close_network(client, "send failed") + return false + } + return true +} + +client_position_worker :: proc(data: rawptr) { + defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) + client := cast(^Client_State)data + position, ok := mpv_get_time_position(client.mpv) + if ok { + client_send_event(client, Playback_Event{kind = .Seek, position = position}) + } else { + fmt.eprintln("client: could not read position after local seek") + } +} + +client_on_pause_change :: proc(client: ^Client_State, paused: bool) { + should_send := false + sync.mutex_lock(&client.mutex) + if client.ignore_initial_pause { + client.ignore_initial_pause = false + client.have_last_pause = true + client.last_pause = paused + } else if client.remote_pause_pending && paused == client.remote_pause_value { + client.remote_pause_pending = false + client.have_last_pause = true + client.last_pause = paused + } else { + client.remote_pause_pending = false + if !client.have_last_pause || paused != client.last_pause { + client.have_last_pause = true + client.last_pause = paused + should_send = true + } + } + sync.mutex_unlock(&client.mutex) + + if should_send { + client_send_event(client, Playback_Event{kind = .Pause, paused = paused}) + } +} + +client_on_mpv_event :: proc(client: ^Client_State, event_name: string, object: json.Object) { + switch event_name { + case "property-change": + name_value, has_name := object["name"] + data_value, has_data := object["data"] + if !has_name || !has_data { + return + } + name, name_ok := name_value.(json.String) + paused, pause_ok := data_value.(json.Boolean) + if name_ok && pause_ok && string(name) == "pause" { + client_on_pause_change(client, bool(paused)) + } + + case "seek": + sync.mutex_lock(&client.mutex) + client.local_seek_pending = true + sync.mutex_unlock(&client.mutex) + + case "playback-restart": + should_query := false + sync.mutex_lock(&client.mutex) + if client.local_seek_pending { + client.local_seek_pending = false + should_query = true + } + sync.mutex_unlock(&client.mutex) + if should_query { + _ = thread.create_and_start_with_data( + rawptr(client), + client_position_worker, + init_context = context, + self_cleanup = true, + ) + } + } +} + +client_on_mpv_disconnect :: proc(client: ^Client_State) { + sync.mutex_lock(&client.mutex) + already_stopping := client.stopping + client.stopping = true + sync.mutex_unlock(&client.mutex) + if !already_stopping { + fmt.eprintln("client: mpv IPC connection closed") + _ = os.process_terminate(client.process) + } +} + +client_apply_remote_event :: proc(client: ^Client_State, event: Playback_Event) -> bool { + switch event.kind { + case .Pause: + sync.mutex_lock(&client.mutex) + client.remote_pause_pending = true + client.remote_pause_value = event.paused + sync.mutex_unlock(&client.mutex) + if !mpv_set_pause(client.mpv, event.paused) { + fmt.eprintln("client: failed to apply remote pause event") + return false + } + return true + case .Seek: + if !mpv_apply_remote_seek(client.mpv, event.position) { + fmt.eprintln("client: failed to apply remote seek safely") + return false + } + return true + } + return false +} + +client_network_reader :: proc(data: rawptr) { + defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) + client := cast(^Client_State)data + framer: Line_Framer + framer_init(&framer) + defer framer_destroy(&framer) + buffer: [2048]byte + reason := "connection closed" + + for { + count, recv_err := net.recv_tcp(client.network_socket, buffer[:]) + if recv_err != nil { + reason = "receive failed" + break + } + if count == 0 { + break + } + frames, frame_err := framer_push(&framer, buffer[:count]) + if frame_err != .None { + destroy_frames(frames) + reason = "oversized message" + break + } + valid := true + for frame in frames { + if len(frame) == 0 { + continue + } + welcome_id, is_welcome, welcome_err := parse_welcome(transmute([]byte)frame) + if is_welcome { + if welcome_err != .None { + fmt.eprintfln("client: invalid welcome message: %s", protocol_error_string(welcome_err)) + reason = "invalid welcome message" + valid = false + break + } + sync.mutex_lock(&client.mutex) + client.client_id = welcome_id + sync.mutex_unlock(&client.mutex) + fmt.printfln("client: assigned client ID %d", welcome_id) + continue + } + _ = welcome_err + + event, protocol_err := parse_protocol_event(transmute([]byte)frame) + if protocol_err != .None { + fmt.eprintfln("client: server sent invalid message: %s", protocol_error_string(protocol_err)) + reason = "invalid server message" + valid = false + break + } + if event.client_id == 0 { + fmt.eprintln("client: server event is missing its triggering client ID") + reason = "event missing client ID" + valid = false + break + } + if !client_apply_remote_event(client, event) { + reason = "could not apply remote event" + valid = false + break + } + log_playback_event("client", event.client_id, event) + } + destroy_frames(frames) + if !valid { + break + } + } + client_close_network(client, reason) +} + +receive_server_welcome :: proc(socket: net.TCP_Socket) -> (client_id: u64, ok: bool) { + buffer: [MAX_FRAME_BYTES+1]byte + length := 0 + for length < len(buffer) { + count, recv_err := net.recv_tcp(socket, buffer[length:length+1]) + if recv_err != nil || count == 0 { + fmt.eprintln("client: server disconnected before assigning a client ID") + return 0, false + } + if buffer[length] == '\n' { + end := length + if end > 0 && buffer[end-1] == '\r' { + end -= 1 + } + matched: bool + protocol_err: Protocol_Error + client_id, matched, protocol_err = parse_welcome(buffer[:end]) + if !matched || protocol_err != .None { + fmt.eprintfln("client: invalid server welcome: %s", protocol_error_string(protocol_err)) + return 0, false + } + return client_id, true + } + length += 1 + } + fmt.eprintln("client: server welcome exceeded the message limit") + return 0, false +} + +start_mpv_process :: proc(options: Options) -> (process: os.Process, ok: bool) { + command := []string{ + options.mpv_path, + "--force-window=yes", + "--idle=yes", + "--pause=yes", + "--start=0", + "--hr-seek=always", + "--keep-open=always", + "--keep-open-pause=yes", + "--input-terminal=no", + "--terminal=no", + fmt.tprintf("--input-ipc-server=%s", options.socket_path), + options.movie_path, + } + process_err: os.Error + process, process_err = os.process_start(os.Process_Desc{ + command = command, + stdout = os.stdout, + stderr = os.stderr, + }) + if process_err != nil { + fmt.eprintfln("client: could not start mpv: %v", process_err) + return {}, false + } + return process, true +} + +wait_for_mpv_socket :: proc(process: os.Process, path: string) -> (posix_fd: int, ok: bool) { + deadline := time.time_add(time.now(), 5*time.Second) + for time.time_to_unix_nano(time.now()) < time.time_to_unix_nano(deadline) { + fd, connected := connect_unix_socket(path) + if connected { + return int(fd), true + } + state, wait_err := os.process_wait(process, timeout = 0) + if wait_err == nil && state.exited { + fmt.eprintln("client: mpv exited before its IPC socket became ready") + return -1, false + } + time.sleep(50*time.Millisecond) + } + fmt.eprintln("client: timed out waiting for mpv IPC socket") + return -1, false +} + +run_client :: proc(options: Options) -> bool { + when ODIN_OS != .Darwin { + fmt.eprintln("client: client mode is supported only on macOS") + return false + } + + network_socket, dial_err := net.dial_tcp_from_hostname_with_port_override(options.host, options.port) + if dial_err != nil { + fmt.eprintfln("client: could not connect to server: %v", dial_err) + return false + } + client_id, welcome_ok := receive_server_welcome(network_socket) + if !welcome_ok { + net.close(network_socket) + return false + } + fmt.printfln("client: assigned client ID %d", client_id) + + process, process_ok := start_mpv_process(options) + if !process_ok { + net.close(network_socket) + return false + } + + fd_value, socket_ok := wait_for_mpv_socket(process, options.socket_path) + if !socket_ok { + _ = os.process_terminate(process) + _, _ = os.process_wait(process) + net.close(network_socket) + return false + } + + client := new(Client_State) + client.client_id = client_id + client.network_socket = network_socket + client.network_active = true + client.process = process + client.ignore_initial_pause = true + mpv := new(Mpv_Connection) + mpv.fd = posix.FD(fd_value) + mpv.client = client + client.mpv = mpv + + mpv_thread := mpv_start_reader(mpv) + if mpv_thread == nil { + fmt.eprintln("client: could not start mpv IPC reader") + mpv_close(mpv) + client_close_network(client, "startup failed") + _ = os.process_terminate(process) + _, _ = os.process_wait(process) + return false + } + if !mpv_observe_pause(mpv) { + fmt.eprintln("client: could not observe mpv pause state") + mpv_close(mpv) + client_close_network(client, "startup failed") + _ = os.process_terminate(process) + _, _ = os.process_wait(process) + thread.join(mpv_thread) + thread.destroy(mpv_thread) + return false + } + + network_thread := thread.create_and_start_with_data( + rawptr(client), + client_network_reader, + init_context = context, + self_cleanup = false, + ) + if network_thread == nil { + fmt.eprintln("client: could not start server reader") + mpv_close(mpv) + client_close_network(client, "startup failed") + _ = os.process_terminate(process) + _, _ = os.process_wait(process) + thread.join(mpv_thread) + thread.destroy(mpv_thread) + return false + } + + fmt.println("client: connected; playback starts paused at 00:00:00") + state: os.Process_State + wait_err: os.Error + for { + state, wait_err = os.process_wait(process, timeout = 100*time.Millisecond) + if wait_err == nil { + break + } + if should_shutdown() { + fmt.println("client: shutdown requested; terminating mpv") + _ = os.process_terminate(process) + state, wait_err = os.process_wait(process, timeout = 2*time.Second) + if wait_err != nil { + _ = os.process_kill(process) + state, wait_err = os.process_wait(process) + } + break + } + if wait_err != os.General_Error.Timeout { + break + } + } + sync.mutex_lock(&client.mutex) + client.stopping = true + sync.mutex_unlock(&client.mutex) + client_close_network(client, "mpv exited") + mpv_close(mpv) + thread.join(network_thread) + thread.destroy(network_thread) + thread.join(mpv_thread) + thread.destroy(mpv_thread) + if wait_err != nil { + fmt.eprintfln("client: waiting for mpv failed: %v", wait_err) + return false + } + if !state.success { + fmt.eprintfln("client: mpv exited with status %d", state.exit_code) + return false + } + return true +} diff --git a/client_test.odin b/client_test.odin new file mode 100644 index 0000000..2e862d9 --- /dev/null +++ b/client_test.odin @@ -0,0 +1,23 @@ +package main + +import "core:testing" + +@(test) +remote_pause_notification_is_suppressed :: proc(t: ^testing.T) { + client := Client_State{ + remote_pause_pending = true, + remote_pause_value = true, + } + client_on_pause_change(&client, true) + testing.expect_value(t, client.remote_pause_pending, false) + testing.expect_value(t, client.have_last_pause, true) + testing.expect_value(t, client.last_pause, true) +} + +@(test) +initial_pause_notification_is_baseline_only :: proc(t: ^testing.T) { + client := Client_State{ignore_initial_pause = true} + client_on_pause_change(&client, true) + testing.expect_value(t, client.ignore_initial_pause, false) + testing.expect_value(t, client.have_last_pause, true) +} diff --git a/framing.odin b/framing.odin new file mode 100644 index 0000000..85c81a6 --- /dev/null +++ b/framing.odin @@ -0,0 +1,70 @@ +package main + +import "core:strings" + +Frame_Error :: enum { + None, + Too_Large, +} + +Line_Framer :: struct { + buffer: [dynamic]byte, +} + +framer_init :: proc(framer: ^Line_Framer, allocator := context.allocator) { + framer.buffer = make([dynamic]byte, 0, 256, allocator) +} + +framer_destroy :: proc(framer: ^Line_Framer) { + delete(framer.buffer) + framer.buffer = nil +} + +destroy_frames :: proc(frames: [dynamic]string) { + for frame in frames { + delete(frame) + } + delete(frames) +} + +framer_push :: proc( + framer: ^Line_Framer, + data: []byte, + allocator := context.allocator, +) -> (frames: [dynamic]string, err: Frame_Error) { + frames = make([dynamic]string, 0, allocator) + append(&framer.buffer, ..data) + + start := 0 + for i := 0; i < len(framer.buffer); i += 1 { + if framer.buffer[i] != '\n' { + if i-start >= MAX_FRAME_BYTES { + return frames, .Too_Large + } + continue + } + + end := i + if end > start && framer.buffer[end-1] == '\r' { + end -= 1 + } + if end-start > MAX_FRAME_BYTES { + return frames, .Too_Large + } + frame := strings.clone(string(framer.buffer[start:end]), allocator) + append(&frames, frame) + start = i+1 + } + + if start > 0 { + remaining := len(framer.buffer)-start + if remaining > 0 { + copy(framer.buffer[:remaining], framer.buffer[start:]) + } + resize(&framer.buffer, remaining) + } + if len(framer.buffer) > MAX_FRAME_BYTES { + return frames, .Too_Large + } + return frames, .None +} diff --git a/framing_test.odin b/framing_test.odin new file mode 100644 index 0000000..50ac930 --- /dev/null +++ b/framing_test.odin @@ -0,0 +1,37 @@ +package main + +import "core:testing" + +@(test) +framer_handles_fragmented_and_coalesced_messages :: proc(t: ^testing.T) { + framer: Line_Framer + framer_init(&framer) + defer framer_destroy(&framer) + + first: string = "one\ntw" + frames, err := framer_push(&framer, transmute([]byte)first) + defer destroy_frames(frames) + testing.expect_value(t, err, Frame_Error.None) + testing.expect_value(t, len(frames), 1) + testing.expect_value(t, frames[0], "one") + + second: string = "o\r\nthree\n" + frames2, err2 := framer_push(&framer, transmute([]byte)second) + defer destroy_frames(frames2) + testing.expect_value(t, err2, Frame_Error.None) + testing.expect_value(t, len(frames2), 2) + testing.expect_value(t, frames2[0], "two") + testing.expect_value(t, frames2[1], "three") +} + +@(test) +framer_rejects_oversized_message :: proc(t: ^testing.T) { + framer: Line_Framer + framer_init(&framer) + defer framer_destroy(&framer) + data := make([]byte, MAX_FRAME_BYTES+1) + defer delete(data) + frames, err := framer_push(&framer, data) + defer destroy_frames(frames) + testing.expect_value(t, err, Frame_Error.Too_Large) +} diff --git a/main.odin b/main.odin new file mode 100644 index 0000000..98aa004 --- /dev/null +++ b/main.odin @@ -0,0 +1,35 @@ +package main + +import "core:fmt" +import "core:os" + +main :: proc() { + install_shutdown_handlers() + options, cli_error := parse_cli(os.args) + if cli_error != "" { + fmt.eprintln("syncplay:", cli_error) + usage() + os.exit(2) + } + if options.command == .Help { + usage() + return + } + + #partial switch options.command { + case .Server: + if !run_server(options) { + os.exit(1) + } + case .Client: + if path_error := validate_client_paths(options); path_error != "" { + fmt.eprintln("syncplay:", path_error) + os.exit(1) + } + if !run_client(options) { + os.exit(1) + } + case: + unreachable() + } +} diff --git a/mpv.odin b/mpv.odin new file mode 100644 index 0000000..9f2900e --- /dev/null +++ b/mpv.odin @@ -0,0 +1,281 @@ +package main + +import "base:runtime" +import "core:encoding/json" +import "core:fmt" +import "core:sync" +import "core:sys/posix" +import "core:thread" +import "core:time" + +MPV_COMMAND_TIMEOUT :: 5*time.Second + +Mpv_Command_Result :: struct { + success: bool, + has_data: bool, + data: f64, +} + +Mpv_Connection :: struct { + fd: posix.FD, + + command_mutex: sync.Mutex, + response_mutex: sync.Mutex, + response_cond: sync.Cond, + next_request_id: i64, + pending_id: i64, + pending_done: bool, + pending_result: Mpv_Command_Result, + closed: bool, + + client: ^Client_State, +} + +connect_unix_socket :: proc(path: string) -> (fd: posix.FD, ok: bool) { + if len(path) == 0 { + return -1, false + } + address: posix.sockaddr_un + if len(path) >= len(address.sun_path) { + return -1, false + } + address.sun_family = .UNIX + when ODIN_OS == .Darwin { + address.sun_len = u8(size_of(address)) + } + for byte_value, index in transmute([]byte)path { + address.sun_path[index] = auto_cast byte_value + } + address.sun_path[len(path)] = 0 + + fd = posix.socket(.UNIX, .STREAM) + if fd < 0 { + return -1, false + } + if posix.connect(fd, (^posix.sockaddr)(&address), posix.socklen_t(size_of(address))) != .OK { + _ = posix.close(fd) + return -1, false + } + return fd, true +} + +mpv_send_all :: proc(fd: posix.FD, message: string) -> bool { + bytes := transmute([]byte)message + written: int = 0 + for written < len(bytes) { + remaining := len(bytes)-written + count := posix.send(fd, raw_data(bytes[written:]), auto_cast remaining, {.NOSIGNAL}) + if count <= 0 { + return false + } + written += int(count) + } + return true +} + +mpv_command_locked :: proc(mpv: ^Mpv_Connection, command_json: string) -> Mpv_Command_Result { + sync.mutex_lock(&mpv.response_mutex) + if mpv.closed { + sync.mutex_unlock(&mpv.response_mutex) + return {} + } + mpv.next_request_id += 1 + request_id := mpv.next_request_id + mpv.pending_id = request_id + mpv.pending_done = false + mpv.pending_result = {} + sync.mutex_unlock(&mpv.response_mutex) + + message := fmt.aprintf("{{\"command\":%s,\"request_id\":%d}}\n", command_json, request_id) + defer delete(message) + if !mpv_send_all(mpv.fd, message) { + return {} + } + + sync.mutex_lock(&mpv.response_mutex) + defer sync.mutex_unlock(&mpv.response_mutex) + deadline_remaining := MPV_COMMAND_TIMEOUT + for !mpv.pending_done && !mpv.closed { + start := time.now() + if !sync.cond_wait_with_timeout(&mpv.response_cond, &mpv.response_mutex, deadline_remaining) { + break + } + elapsed := time.since(start) + if elapsed >= deadline_remaining { + break + } + deadline_remaining -= elapsed + } + if !mpv.pending_done || mpv.pending_id != request_id { + return {} + } + return mpv.pending_result +} + +mpv_command :: proc(mpv: ^Mpv_Connection, command_json: string) -> Mpv_Command_Result { + sync.mutex_lock(&mpv.command_mutex) + defer sync.mutex_unlock(&mpv.command_mutex) + return mpv_command_locked(mpv, command_json) +} + +mpv_apply_remote_seek :: proc(mpv: ^Mpv_Connection, position: f64) -> bool { + sync.mutex_lock(&mpv.command_mutex) + defer sync.mutex_unlock(&mpv.command_mutex) + + disabled := mpv_command_locked(mpv, `["disable_event","seek"]`) + if !disabled.success { + return false + } + reenabled := false + defer if !reenabled { + result := mpv_command_locked(mpv, `["enable_event","seek"]`) + if !result.success { + fmt.eprintln("client: failed to re-enable mpv seek events") + } + } + + seek_command := fmt.aprintf(`["seek",%.6f,"absolute+exact"]`, position) + defer delete(seek_command) + seek_result := mpv_command_locked(mpv, seek_command) + enable_result := mpv_command_locked(mpv, `["enable_event","seek"]`) + reenabled = enable_result.success + return seek_result.success && enable_result.success +} + +mpv_set_pause :: proc(mpv: ^Mpv_Connection, paused: bool) -> bool { + command := fmt.aprintf( + `["set_property","pause",%s]`, + "true" if paused else "false", + ) + defer delete(command) + return mpv_command(mpv, command).success +} + +mpv_get_time_position :: proc(mpv: ^Mpv_Connection) -> (f64, bool) { + result := mpv_command(mpv, `["get_property","time-pos"]`) + return result.data, result.success && result.has_data +} + +mpv_observe_pause :: proc(mpv: ^Mpv_Connection) -> bool { + return mpv_command(mpv, `["observe_property",1,"pause"]`).success +} + +mpv_parse_response :: proc(mpv: ^Mpv_Connection, object: json.Object) -> bool { + id_value, has_id := object["request_id"] + if !has_id { + return false + } + id, id_ok := id_value.(json.Integer) + if !id_ok { + return false + } + error_value, has_error := object["error"] + if !has_error { + return false + } + error_name, error_ok := error_value.(json.String) + if !error_ok { + return false + } + + result := Mpv_Command_Result{success = string(error_name) == "success"} + if data_value, has_data := object["data"]; has_data { + #partial switch value in data_value { + case json.Integer: + result.has_data = true + result.data = f64(value) + case json.Float: + result.has_data = true + result.data = f64(value) + } + } + + sync.mutex_lock(&mpv.response_mutex) + if i64(id) == mpv.pending_id { + mpv.pending_result = result + mpv.pending_done = true + sync.cond_signal(&mpv.response_cond) + } + sync.mutex_unlock(&mpv.response_mutex) + return true +} + +mpv_handle_line :: proc(mpv: ^Mpv_Connection, line: string) { + root: json.Value + if json.unmarshal(transmute([]byte)line, &root, spec = .JSON) != nil { + fmt.eprintln("client: received malformed JSON from mpv") + return + } + defer json.destroy_value(root) + object, ok := root.(json.Object) + if !ok { + return + } + if mpv_parse_response(mpv, object) { + return + } + + event_value, has_event := object["event"] + if !has_event { + return + } + event_name, event_ok := event_value.(json.String) + if !event_ok { + return + } + if mpv.client != nil { + client_on_mpv_event(mpv.client, string(event_name), object) + } +} + +mpv_reader_loop :: proc(data: rawptr) { + defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) + mpv := cast(^Mpv_Connection)data + framer: Line_Framer + framer_init(&framer) + defer framer_destroy(&framer) + buffer: [2048]byte + + for { + count := posix.recv(mpv.fd, raw_data(buffer[:]), len(buffer), {}) + if count <= 0 { + break + } + frames, frame_err := framer_push(&framer, buffer[:int(count)]) + if frame_err != .None { + destroy_frames(frames) + break + } + for frame in frames { + if len(frame) > 0 { + mpv_handle_line(mpv, frame) + } + } + destroy_frames(frames) + } + + sync.mutex_lock(&mpv.response_mutex) + mpv.closed = true + sync.cond_broadcast(&mpv.response_cond) + sync.mutex_unlock(&mpv.response_mutex) + if mpv.client != nil { + client_on_mpv_disconnect(mpv.client) + } +} + +mpv_start_reader :: proc(mpv: ^Mpv_Connection) -> ^thread.Thread { + return thread.create_and_start_with_data( + rawptr(mpv), + mpv_reader_loop, + init_context = context, + self_cleanup = false, + ) +} + +mpv_close :: proc(mpv: ^Mpv_Connection) { + if mpv.fd >= 0 { + _ = posix.shutdown(mpv.fd, .RDWR) + _ = posix.close(mpv.fd) + mpv.fd = -1 + } +} diff --git a/mpv_test.odin b/mpv_test.odin new file mode 100644 index 0000000..973684b --- /dev/null +++ b/mpv_test.odin @@ -0,0 +1,100 @@ +package main + +import "base:runtime" +import "core:fmt" +import "core:strings" +import "core:sys/posix" +import "core:testing" +import "core:thread" + +Fake_Mpv :: struct { + fd: posix.FD, + received: [3]string, + fail_seek: bool, +} + +fake_mpv_read_line :: proc(fd: posix.FD, buffer: []byte) -> (line: string, ok: bool) { + length := 0 + for length < len(buffer) { + count := posix.recv(fd, raw_data(buffer[length:]), 1, {}) + if count != 1 { + return "", false + } + if buffer[length] == '\n' { + return string(buffer[:length]), true + } + length += 1 + } + return "", false +} + +fake_mpv_loop :: proc(data: rawptr) { + defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) + fake := cast(^Fake_Mpv)data + buffer: [512]byte + for index in 0..<3 { + line, ok := fake_mpv_read_line(fake.fd, buffer[:]) + if !ok { + return + } + fake.received[index] = strings.clone(line) + error_name := "failure" if fake.fail_seek && index == 1 else "success" + response := fmt.aprintf( + "{{\"error\":\"%s\",\"request_id\":%d}}\n", + error_name, + index+1, + ) + _ = mpv_send_all(fake.fd, response) + delete(response) + } +} + +run_fake_seek_transaction :: proc(t: ^testing.T, fail_seek: bool) -> (ok: bool, fake: Fake_Mpv) { + fds: [2]posix.FD + testing.expect_value(t, posix.socketpair(.UNIX, .STREAM, .IP, &fds), posix.result.OK) + + mpv := Mpv_Connection{fd = fds[0]} + reader := mpv_start_reader(&mpv) + testing.expect(t, reader != nil) + fake = Fake_Mpv{fd = fds[1], fail_seek = fail_seek} + fake_thread := thread.create_and_start_with_data( + rawptr(&fake), + fake_mpv_loop, + init_context = context, + self_cleanup = false, + ) + testing.expect(t, fake_thread != nil) + + ok = mpv_apply_remote_seek(&mpv, 42.5) + thread.join(fake_thread) + thread.destroy(fake_thread) + mpv_close(&mpv) + thread.join(reader) + thread.destroy(reader) + _ = posix.close(fake.fd) + return +} + +@(test) +remote_seek_masks_and_restores_seek_events :: proc(t: ^testing.T) { + ok, fake := run_fake_seek_transaction(t, false) + testing.expect(t, ok) + + testing.expect(t, strings.contains(fake.received[0], `"disable_event","seek"`)) + testing.expect(t, strings.contains(fake.received[1], `"seek",42.500000,"absolute+exact"`)) + testing.expect(t, strings.contains(fake.received[2], `"enable_event","seek"`)) + for line in fake.received { + delete(line) + } +} + +@(test) +remote_seek_restores_events_after_seek_failure :: proc(t: ^testing.T) { + ok, fake := run_fake_seek_transaction(t, true) + testing.expect(t, !ok) + testing.expect(t, strings.contains(fake.received[0], `"disable_event","seek"`)) + testing.expect(t, strings.contains(fake.received[2], `"enable_event","seek"`)) + for line in fake.received { + delete(line) + } +} diff --git a/protocol.odin b/protocol.odin new file mode 100644 index 0000000..7121890 --- /dev/null +++ b/protocol.odin @@ -0,0 +1,278 @@ +package main + +import "core:encoding/json" +import "core:fmt" +import "core:math" +import "core:strings" + +PROTOCOL_VERSION :: 1 +MAX_FRAME_BYTES :: 4096 + +Event_Kind :: enum { + Pause, + Seek, +} + +Playback_Event :: struct { + kind: Event_Kind, + paused: bool, + position: f64, + client_id: u64, +} + +Protocol_Error :: enum { + None, + Invalid_JSON, + Expected_Object, + Missing_Version, + Unsupported_Version, + Missing_Type, + Unknown_Type, + Missing_Paused, + Invalid_Paused, + Missing_Position, + Invalid_Position, + Missing_Client_ID, + Invalid_Client_ID, +} + +protocol_error_string :: proc(err: Protocol_Error) -> string { + switch err { + case .None: return "no error" + case .Invalid_JSON: return "invalid JSON" + case .Expected_Object: return "expected a JSON object" + case .Missing_Version: return "missing integer version" + case .Unsupported_Version: return "unsupported protocol version" + case .Missing_Type: return "missing string type" + case .Unknown_Type: return "unknown event type" + case .Missing_Paused: return "missing paused field" + case .Invalid_Paused: return "paused must be a boolean" + case .Missing_Position: return "missing position field" + case .Invalid_Position: return "position must be a finite non-negative number" + case .Missing_Client_ID: return "missing client_id field" + case .Invalid_Client_ID: return "client_id must be a positive integer" + } + return "unknown protocol error" +} + +parse_protocol_event :: proc(data: []byte) -> (event: Playback_Event, err: Protocol_Error) { + root: json.Value + if json.unmarshal(data, &root, spec = .JSON) != nil { + return {}, .Invalid_JSON + } + defer json.destroy_value(root) + + object, ok := root.(json.Object) + if !ok { + return {}, .Expected_Object + } + + version_value, found := object["version"] + if !found { + return {}, .Missing_Version + } + version, version_ok := version_value.(json.Integer) + if !version_ok { + return {}, .Missing_Version + } + if version != PROTOCOL_VERSION { + return {}, .Unsupported_Version + } + + type_value, type_found := object["type"] + if !type_found { + return {}, .Missing_Type + } + type_name, type_ok := type_value.(json.String) + if !type_ok { + return {}, .Missing_Type + } + + client_id: u64 + if client_id_value, client_id_found := object["client_id"]; client_id_found { + parsed_id, client_id_ok := client_id_value.(json.Integer) + if !client_id_ok || parsed_id <= 0 { + return {}, .Invalid_Client_ID + } + client_id = u64(parsed_id) + } + + switch string(type_name) { + case "pause": + paused_value, paused_found := object["paused"] + if !paused_found { + return {}, .Missing_Paused + } + paused, paused_ok := paused_value.(json.Boolean) + if !paused_ok { + return {}, .Invalid_Paused + } + return Playback_Event{kind = .Pause, paused = bool(paused), client_id = client_id}, .None + + case "seek": + position_value, position_found := object["position"] + if !position_found { + return {}, .Missing_Position + } + position: f64 + #partial switch value in position_value { + case json.Integer: + position = f64(value) + case json.Float: + position = f64(value) + case: + return {}, .Invalid_Position + } + if position < 0 || math.is_nan(position) || math.is_inf(position) { + return {}, .Invalid_Position + } + return Playback_Event{kind = .Seek, position = position, client_id = client_id}, .None + } + + return {}, .Unknown_Type +} + +encode_protocol_event :: proc(event: Playback_Event, allocator := context.allocator) -> (line: string) { + switch event.kind { + case .Pause: + if event.client_id > 0 { + return fmt.aprintf( + "{{\"version\":%d,\"type\":\"pause\",\"paused\":%s,\"client_id\":%d}}\n", + PROTOCOL_VERSION, + "true" if event.paused else "false", + event.client_id, + allocator = allocator, + ) + } + return fmt.aprintf( + "{{\"version\":%d,\"type\":\"pause\",\"paused\":%s}}\n", + PROTOCOL_VERSION, + "true" if event.paused else "false", + allocator = allocator, + ) + case .Seek: + if event.client_id > 0 { + return fmt.aprintf( + "{{\"version\":%d,\"type\":\"seek\",\"position\":%.6f,\"client_id\":%d}}\n", + PROTOCOL_VERSION, + event.position, + event.client_id, + allocator = allocator, + ) + } + return fmt.aprintf( + "{{\"version\":%d,\"type\":\"seek\",\"position\":%.6f}}\n", + PROTOCOL_VERSION, + event.position, + allocator = allocator, + ) + } + return strings.clone("", allocator) +} + +encode_welcome :: proc(client_id: u64, allocator := context.allocator) -> string { + return fmt.aprintf( + "{{\"version\":%d,\"type\":\"welcome\",\"client_id\":%d}}\n", + PROTOCOL_VERSION, + client_id, + allocator = allocator, + ) +} + +parse_welcome :: proc(data: []byte) -> (client_id: u64, matched: bool, err: Protocol_Error) { + root: json.Value + if json.unmarshal(data, &root, spec = .JSON) != nil { + return 0, false, .Invalid_JSON + } + defer json.destroy_value(root) + object, object_ok := root.(json.Object) + if !object_ok { + return 0, false, .Expected_Object + } + type_value, type_found := object["type"] + if !type_found { + return 0, false, .Missing_Type + } + type_name, type_ok := type_value.(json.String) + if !type_ok { + return 0, false, .Missing_Type + } + if string(type_name) != "welcome" { + return 0, false, .None + } + matched = true + version_value, version_found := object["version"] + if !version_found { + return 0, true, .Missing_Version + } + version, version_ok := version_value.(json.Integer) + if !version_ok { + return 0, true, .Missing_Version + } + if version != PROTOCOL_VERSION { + return 0, true, .Unsupported_Version + } + client_id_value, client_id_found := object["client_id"] + if !client_id_found { + return 0, true, .Missing_Client_ID + } + parsed_id, client_id_ok := client_id_value.(json.Integer) + if !client_id_ok || parsed_id <= 0 { + return 0, true, .Invalid_Client_ID + } + return u64(parsed_id), true, .None +} + +format_playback_log :: proc( + scope: string, + client_id: u64, + event: Playback_Event, + is_local := false, + allocator := context.allocator, +) -> string { + if is_local { + switch event.kind { + case .Pause: + return fmt.aprintf( + "%s: client %d (you) %s", + scope, + client_id, + "paused" if event.paused else "played", + allocator = allocator, + ) + case .Seek: + return fmt.aprintf( + "%s: client %d (you) sought to %.3f seconds", + scope, + client_id, + event.position, + allocator = allocator, + ) + } + } + switch event.kind { + case .Pause: + return fmt.aprintf( + "%s: client %d %s", + scope, + client_id, + "paused" if event.paused else "played", + allocator = allocator, + ) + case .Seek: + return fmt.aprintf( + "%s: client %d sought to %.3f seconds", + scope, + client_id, + event.position, + allocator = allocator, + ) + } + return strings.clone("", allocator) +} + +log_playback_event :: proc(scope: string, client_id: u64, event: Playback_Event, is_local := false) { + line := format_playback_log(scope, client_id, event, is_local) + defer delete(line) + fmt.println(line) +} diff --git a/protocol_test.odin b/protocol_test.odin new file mode 100644 index 0000000..7c9996b --- /dev/null +++ b/protocol_test.odin @@ -0,0 +1,73 @@ +package main + +import "core:strings" +import "core:testing" + +@(test) +protocol_round_trip_pause :: proc(t: ^testing.T) { + line := encode_protocol_event(Playback_Event{kind = .Pause, paused = true}) + defer delete(line) + event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line)) + testing.expect_value(t, err, Protocol_Error.None) + testing.expect_value(t, event.kind, Event_Kind.Pause) + testing.expect_value(t, event.paused, true) +} + +@(test) +protocol_round_trip_seek :: proc(t: ^testing.T) { + line := encode_protocol_event(Playback_Event{kind = .Seek, position = 123.456}) + defer delete(line) + event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line)) + testing.expect_value(t, err, Protocol_Error.None) + testing.expect_value(t, event.kind, Event_Kind.Seek) + testing.expect(t, event.position > 123.455 && event.position < 123.457) +} + +@(test) +protocol_rejects_invalid_messages :: proc(t: ^testing.T) { + missing_pause: string = `{"version":1,"type":"pause"}` + bad_version: string = `{"version":2,"type":"pause","paused":true}` + bad_position: string = `{"version":1,"type":"seek","position":-1}` + _, err := parse_protocol_event(transmute([]byte)missing_pause) + testing.expect_value(t, err, Protocol_Error.Missing_Paused) + _, err = parse_protocol_event(transmute([]byte)bad_version) + testing.expect_value(t, err, Protocol_Error.Unsupported_Version) + _, err = parse_protocol_event(transmute([]byte)bad_position) + testing.expect_value(t, err, Protocol_Error.Invalid_Position) +} + +@(test) +protocol_round_trip_sourced_event :: proc(t: ^testing.T) { + line := encode_protocol_event(Playback_Event{ + kind = .Seek, + position = 9.25, + client_id = 7, + }) + defer delete(line) + event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line)) + testing.expect_value(t, err, Protocol_Error.None) + testing.expect_value(t, event.client_id, u64(7)) +} + +@(test) +protocol_parses_welcome :: proc(t: ^testing.T) { + line := encode_welcome(12) + defer delete(line) + client_id, matched, err := parse_welcome(transmute([]byte)strings.trim_space(line)) + testing.expect(t, matched) + testing.expect_value(t, err, Protocol_Error.None) + testing.expect_value(t, client_id, u64(12)) +} + +@(test) +playback_logs_identify_client_and_action :: proc(t: ^testing.T) { + paused := format_playback_log("server", 2, Playback_Event{kind = .Pause, paused = true}) + defer delete(paused) + played := format_playback_log("client", 3, Playback_Event{kind = .Pause}, is_local = true) + defer delete(played) + sought := format_playback_log("client", 4, Playback_Event{kind = .Seek, position = 65.125}) + defer delete(sought) + testing.expect_value(t, paused, "server: client 2 paused") + testing.expect_value(t, played, "client: client 3 (you) played") + testing.expect_value(t, sought, "client: client 4 sought to 65.125 seconds") +} diff --git a/server.odin b/server.odin new file mode 100644 index 0000000..70814fb --- /dev/null +++ b/server.odin @@ -0,0 +1,196 @@ +package main + +import "base:runtime" +import "core:fmt" +import "core:net" +import "core:sync" +import "core:thread" +import "core:sys/posix" + +MAX_SERVER_CLIENTS :: 16 + +Server_State :: struct { + mutex: sync.Mutex, + clients: [MAX_SERVER_CLIENTS]Server_Client, + next_id: u64, +} + +Server_Client :: struct { + state: ^Server_State, + socket: net.TCP_Socket, + endpoint: net.Endpoint, + id: u64, + active: bool, +} + +server_disconnect_locked :: proc(client: ^Server_Client, reason: string) { + if !client.active { + return + } + client.active = false + _ = net.shutdown(client.socket, .Both) + net.close(client.socket) + fmt.printfln("server: client %d disconnected (%s)", client.id, reason) +} + +server_broadcast :: proc(sender: ^Server_Client, event: Playback_Event) { + state := sender.state + sync.mutex_lock(&state.mutex) + defer sync.mutex_unlock(&state.mutex) + if !sender.active { + return + } + outbound_event := event + outbound_event.client_id = sender.id + log_playback_event("server", sender.id, outbound_event) + line := encode_protocol_event(outbound_event) + defer delete(line) + + for &client in state.clients { + if !client.active || client.id == sender.id { + continue + } + written, send_err := net.send_tcp(client.socket, transmute([]byte)line) + if send_err != nil || written != len(line) { + server_disconnect_locked(&client, "send failed") + } + } +} + +server_client_loop :: proc(data: rawptr) { + defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) + client := cast(^Server_Client)data + framer: Line_Framer + framer_init(&framer) + defer framer_destroy(&framer) + buf: [2048]byte + disconnect_reason := "connection closed" + + for { + count, recv_err := net.recv_tcp(client.socket, buf[:]) + if recv_err != nil { + disconnect_reason = "receive failed" + break + } + if count == 0 { + break + } + + frames, frame_err := framer_push(&framer, buf[:count]) + if frame_err != .None { + destroy_frames(frames) + disconnect_reason = "oversized message" + break + } + + valid := true + for frame in frames { + if len(frame) == 0 { + continue + } + event, protocol_err := parse_protocol_event(transmute([]byte)frame) + if protocol_err != .None { + fmt.printfln( + "server: client %d sent invalid message: %s", + client.id, + protocol_error_string(protocol_err), + ) + disconnect_reason = "invalid message" + valid = false + break + } + server_broadcast(client, event) + } + destroy_frames(frames) + if !valid { + break + } + } + + sync.mutex_lock(&client.state.mutex) + server_disconnect_locked(client, disconnect_reason) + sync.mutex_unlock(&client.state.mutex) +} + +run_server :: proc(options: Options) -> bool { + address, ok := net.parse_ip4_address(options.host) + if !ok { + fmt.eprintfln("server: host must be an IPv4 address: %s", options.host) + return false + } + listener, listen_err := net.listen_tcp(net.Endpoint{address = address, port = options.port}) + if listen_err != nil { + fmt.eprintfln("server: could not listen on %s:%d: %v", options.host, options.port, listen_err) + return false + } + register_server_listener(posix.FD(net.Socket(listener))) + defer clear_server_listener() + + state := new(Server_State) + fmt.printfln("server: listening on %s:%d", options.host, options.port) + + for { + socket, endpoint, accept_err := net.accept_tcp(listener) + if accept_err != nil { + if should_shutdown() { + fmt.println("server: shutdown requested") + return true + } + fmt.eprintfln("server: accept failed: %v", accept_err) + net.close(listener) + return false + } + + client: ^Server_Client + client_ready := false + sync.mutex_lock(&state.mutex) + for &candidate in state.clients { + if !candidate.active { + client = &candidate + break + } + } + if client != nil { + state.next_id += 1 + client^ = Server_Client{ + state = state, + socket = socket, + endpoint = endpoint, + id = state.next_id, + active = true, + } + welcome := encode_welcome(client.id) + welcome_length := len(welcome) + written, welcome_err := net.send_tcp(client.socket, transmute([]byte)welcome) + delete(welcome) + if welcome_err == nil && written == welcome_length { + client_ready = true + } else { + server_disconnect_locked(client, "could not send welcome") + } + } + sync.mutex_unlock(&state.mutex) + + if client == nil { + fmt.eprintln("server: connection rejected: client limit reached") + net.close(socket) + continue + } + if !client_ready { + continue + } + + remote := net.endpoint_to_string(endpoint) + fmt.printfln("server: client %d connected from %s", client.id, remote) + if thread.create_and_start_with_data( + rawptr(client), + server_client_loop, + init_context = context, + self_cleanup = true, + ) == nil { + sync.mutex_lock(&state.mutex) + server_disconnect_locked(client, "could not start client thread") + sync.mutex_unlock(&state.mutex) + } + } +} diff --git a/shutdown.odin b/shutdown.odin new file mode 100644 index 0000000..0b80e56 --- /dev/null +++ b/shutdown.odin @@ -0,0 +1,34 @@ +package main + +import "base:intrinsics" +import "core:c" +import "core:c/libc" +import "core:sys/posix" + +shutdown_requested: libc.sig_atomic_t +server_listener_fd: libc.sig_atomic_t = -1 + +shutdown_signal_handler :: proc "c" (signal: c.int) { + intrinsics.atomic_store(&shutdown_requested, 1) + fd := intrinsics.atomic_exchange(&server_listener_fd, libc.sig_atomic_t(-1)) + if fd >= 0 { + _ = posix.close(posix.FD(fd)) + } +} + +register_server_listener :: proc(fd: posix.FD) { + intrinsics.atomic_store(&server_listener_fd, libc.sig_atomic_t(fd)) +} + +clear_server_listener :: proc() { + intrinsics.atomic_store(&server_listener_fd, libc.sig_atomic_t(-1)) +} + +install_shutdown_handlers :: proc() { + libc.signal(libc.SIGINT, shutdown_signal_handler) + libc.signal(libc.SIGTERM, shutdown_signal_handler) +} + +should_shutdown :: proc() -> bool { + return intrinsics.atomic_load(&shutdown_requested) == 1 +}