diff --git a/docs/integration-guide.md b/docs/integration-guide.md index a7b57148..0c8f1870 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -51,7 +51,7 @@ SendspinClient client(std::move(config)); ## Step 2: Add Roles -Add only the roles your application needs. All roles must be added before calling `start_server()`. +Add only the roles your application needs. All roles must be added before calling `start()`. ### Player Role (Audio Playback) @@ -499,7 +499,7 @@ struct MyClientListener : SendspinClientListener { ## Step 5: Wire Everything Together -Listeners and providers are set as raw pointers. They must outlive the client. +Listeners and providers are set as raw pointers. They must stay alive for as long as the client can call them: until `stop()` returns, or until the client is destroyed if `stop()` is never called. The destructor itself never invokes a listener (see [Stopping and Restarting](#stopping-and-restarting)). ```cpp MyPlayerListener player_listener; @@ -520,9 +520,10 @@ client.set_persistence_provider(&persistence_provider); // Optional ## Step 6: Start and Run ```cpp -// Start the WebSocket server and sync task. +// Start the role threads and arm the WebSocket server (it comes up on the first loop() tick +// after the network provider reports ready). // Task priorities and PSRAM settings are taken from SendspinClientConfig. -if (!client.start_server()) { +if (!client.start()) { // Handle failure return 1; } @@ -537,10 +538,30 @@ while (running) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } -// Clean shutdown -client.disconnect(SendspinGoodbyeReason::SHUTDOWN); +// Clean shutdown: goodbye every peer, tear everything down, deliver the clear callbacks. +client.stop(); ``` +`start_server()` is a deprecated alias of `start()`. + +## Stopping and Restarting + +`stop()` is synchronous: when it returns the client is fully stopped. It sends a `client/goodbye` (reason `shutdown`) to every peer, waits up to a short bound (50 ms per peer) for those sends to complete, then closes the server and every connection regardless, joins the role threads, resets every role, and delivers the roles' clear callbacks (`on_stream_end()`, `on_image_clear()`, `on_visualizer_stream_end()`, `on_metadata_clear()`, `on_controller_state_clear()`, `on_color_clear()`) before returning. It is a no-op on a stopped client. `is_started()` reports the state, and `loop()` is a no-op while stopped. + +Restarting is `start()` again; start, stop, and start again can be repeated indefinitely, and a restarted client begins with no connection, no group state, and no role state from before the stop. + +`stop()` may block, but the wait is bounded. Besides the goodbye bound it includes: + +- The transports' own close. The host server joins every accepted connection thread; a WebSocket peer completes its close handshake within about 300 ms, but a raw socket that connected and never completed the upgrade holds the join for the full 3 s handshake timeout. The ESP server waits for the httpd task to exit, which polls at 100 ms and first finishes any queued send, which can take up to httpd's send timeout for a peer that has stopped reading. +- An outbound `connect_to()` connection's transport stop, which is synchronous (`esp_websocket_client_stop()` / `ix::WebSocket::stop()`). +- A listener callback already running on a role thread: the join cannot interrupt it. `on_audio_write()` is bounded by its `timeout_ms`; `on_image_decode()` has no bound. + +Listener callbacks fire from inside `stop()`, after every role and the group state have been reset, so a callback that reads the client through its getters sees the stopped state. One that calls `start()` gets `false` and starts nothing; one that calls `stop()`, `connect_to()`, or `disconnect()` is ignored. `is_started()` reads `false` throughout and is safe to call from any thread. Call `stop()` only from the main loop thread: from a role-thread callback it would join the calling thread. + +`on_request_high_performance()` and `on_release_high_performance()` can fire while the client holds an internal lock, so their bodies must only toggle the platform networking mode and must not call any client or role method. + +Destroying a running client performs the transport half of `stop()` (goodbye, bounded wait, close, join) and dispatches no teardown or clear callback. Role-thread callbacks (`on_audio_write()`, `on_image_decode()`, visualizer deliveries) can still run until the destructor joins their role, so listeners must outlive the client as described in Step 5. Call `stop()` first when the clear callbacks matter. + ## Sending Commands If you added the controller role, use it to send playback commands. `send_command` takes a `ClientCommandControllerObject`, built with designated initializers - set only the field the command uses: @@ -700,7 +721,7 @@ int main() { player.set_listener(&player_listener); client.set_network_provider(&network); - client.start_server(); + client.start(); while (true) { client.loop(); diff --git a/docs/internals.md b/docs/internals.md index 29e0a9e4..bbea5db3 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -47,21 +47,22 @@ On host builds, `platform_configure_thread()` is a no-op; threads use OS default 1. `SyncTask::start()` configures the thread and spawns it. 2. The caller blocks until the thread reaches IDLE state (`TASK_IDLE` event flag) or exits early due to an allocation failure (`TASK_STOPPED`). -3. The thread runs a persistent outer loop for the lifetime of the client. -4. `SyncTask::stop()` sets `COMMAND_STOP`, wakes the ring buffer receive via `wake_receiver()`, and joins the thread. Called from `SyncTask`'s destructor, which is triggered by `sync_task_.reset()` in `PlayerRole::Impl`'s destructor. +3. The thread runs a persistent outer loop for one started session, until `stop()`. +4. `SyncTask::stop()` sets `COMMAND_STOP`, wakes the ring buffer receive via `wake_receiver()`, and joins the thread; after the join it clears `TASK_RUNNING` (a stop mid-stream leaves it set, and the player's sync-idle gate must read a stopped task as idle) and resets the encoded ring buffer, so a later `start()` begins with an empty ring. Called from `PlayerRole::Impl::stop()` (`SendspinClient::stop()` and the client destructor) and from `SyncTask`'s destructor, which is triggered by `sync_task_.reset()` in `PlayerRole::Impl`'s destructor. +5. `SyncTask::start()` clears every command and state flag before spawning, so a restart after `stop()` inherits nothing from the previous thread. **Visualizer drain** (`src/visualizer_role.cpp`): 1. `VisualizerRole::Impl::start()` spawns the drain thread. 2. The thread blocks on ring buffer receives; commands interrupt the receive immediately via `wake_receiver()`. The 5 s receive timeout is only a fallback against a missed wake. -3. `VisualizerRole::Impl` destructor sets `COMMAND_STOP`, wakes the ring buffer receive, and joins. +3. `VisualizerRole::Impl::stop()` (from `SendspinClient::stop()`, the client destructor, and the `Impl` destructor) sets `COMMAND_STOP`, wakes the ring buffer receive, joins, and then flushes the ring buffer: with the thread joined it is the ring's only consumer, and a restart must not deliver the previous session's frames. `start()` clears `COMMAND_STOP`, `COMMAND_FLUSH`, and `COMMAND_CLEAR` before spawning, since `cleanup()` on a stopped role leaves a flush flagged. **Artwork decode** (`src/artwork_role.cpp`): 1. `ArtworkRole::Impl::start()` spawns the decode thread. 2. The thread blocks on notification queue receives; commands interrupt the receive immediately via `wake_receiver()`. The 5 s receive timeout is only a fallback against a missed wake. 3. On notification: calls `on_image_decode()`, then merges an `ArtworkDisplayUpdate` (the slot's server display timestamp plus the `stream_epoch` it was decoded under) into the `ArtworkRole::Impl::EventState::display_slot` `InboxSlot` via `merge_artwork_display_update`. The main loop's `ArtworkRole::Impl::drain_events()` folds the taken update into its main-thread-only `held_display_*` state and fires `on_image_display()` once the timestamp is reached. Latest-wins per slot: if a newer frame's timestamp overwrites the pending one before the main loop takes it, only the newer display fires; the per-slot epoch lets the deadline sweep drop a display whose stream was replaced after the hand-off. -4. `ArtworkRole::Impl` destructor sets `COMMAND_STOP`, wakes the queue receive, and joins. +4. `ArtworkRole::Impl::stop()` (from `SendspinClient::stop()`, the client destructor, and the `Impl` destructor) sets `COMMAND_STOP`, wakes the queue receive, joins, and then resets the notification queue so a restart does not decode the previous session's images. `start()` clears `COMMAND_STOP` before spawning. **Destruction order** matters because external audio callbacks may still reference the sync task. `PlayerRole::Impl`'s destructor resets the sync task first (`sync_task_.reset()`) before tearing down anything else, so the thread is fully joined before any shared state is destroyed. @@ -202,7 +203,7 @@ The bump arena suits ArduinoJson's allocation pattern: during a parse the varian ### Main Loop Processing -`SendspinClient::loop()` (`src/client.cpp`) runs the following steps **in order** on each tick: +`SendspinClient::loop()` (`src/client.cpp`) is a no-op while the client is stopped (a stopped client has no connections or threads, and the manager loop must not restart the WebSocket server). While started it runs the following steps **in order** on each tick; steps 3 onward are `SendspinClient::drain_inbox()`, which `stop()` also calls once so the clear callbacks are delivered synchronously: ```api 1. connection_manager_->loop() (sections gated on lock-free atomic hints - see below) @@ -451,6 +452,56 @@ When a connection is lost (`on_connection_lost`): `disable_message_dispatch()` is the first step because it's an atomic flag that the network thread checks before invoking any callback. This prevents stale messages from a dead connection from racing into freshly-reset role queues. +### Client Start and Stop + +`SendspinClient::start()` loads persisted state, starts the threaded roles (player sync task, visualizer drain, artwork decode; a failure part-way stops the ones that did start), and calls `ConnectionManager::start()`, which opens admission (`accepting_`) and creates the `SendspinWsServer` on first use. The server itself is started by the manager's `loop()` once the network provider reports ready, so `is_started()` means "running", not "listening". + +`SendspinClient::stop()` is synchronous and ordered so that every producer is gone before any state is reset. The client's lifecycle is one atomic `lifecycle_` field (`STOPPED`, `RUNNING`, `STOPPING`); `is_started()` reads it from any thread. + +```api +0. lifecycle_ = STOPPING (is_started() reads false; loop() is a no-op; start() is refused and + stop()/connect_to()/disconnect() are ignored from here on, so a callback fired below cannot + recurse into the teardown) +1. VisualizerRole/ArtworkRole::Impl::signal_stop(): set COMMAND_STOP and wake, no join, so a + slow on_image_decode() or a parked drain exits while the transports close. The player is + not signalled yet: a network thread blocked on its ring (write_audio_chunk) needs the sync + task alive until the network threads are gone +2. ConnectionManager::stop(SHUTDOWN) + ├─ Under conn_ptr_mutex_: accepting_ = false; disable_message_dispatch() on every managed + │ connection; move the current slot, the nursery, and the deferred releases out; clear the + │ hello retries + ├─ Outside the lock: conn->disconnect(SHUTDOWN, completion) on each, completion counted by a + │ shared GoodbyeWait; wait up to GOODBYE_FLUSH_TIMEOUT_MS (50 ms) per goodbye for the count + │ to reach zero (the ESP httpd worker hands the frames to lwIP one at a time) + ├─ ws_server_->stop() regardless (host: joins every accepted connection thread, a WebSocket + │ peer within its ~300 ms close handshake and a raw never-upgraded socket within the 3 s + │ WS_HANDSHAKE_TIMEOUT_SECS; ESP: httpd_stop(), which runs queued sends first, + │ then every session's close_fn and ctx free_fn, polling at 100 ms) + └─ take_pending_events(): move the pending connected/disconnect event queues out under + conn_mutex_; every moved-out shared_ptr is released outside the locks (an outbound + connection's destructor stops its transport synchronously) +3. Role threads: PlayerRole/VisualizerRole/ArtworkRole::Impl::stop() join, then each discards + its ring/queue content (sole consumer after the join) +4. cleanup_connection_state() (the same reset a lost connection triggers, including the group + slot), then group_state_ and state_ are reset +5. drain_inbox() delivers the CLEARED / STREAM_END callbacks step 4 queued. Every getter already + reports the stopped state, so a callback that reads the client sees what a caller sees once + stop() returns +6. lifecycle_ = STOPPED +``` + +The goodbye completion is best-effort: on ESP a session that closes before its queued worker runs, or whose `weak_ptr` no longer resolves, never reports, which is why the wait is bounded rather than exact. The `GoodbyeWait` record is held by `shared_ptr` and captured by value in each completion, so a completion that runs late on a transport thread touches nothing `stop()` owns. A peer delivered by the ws_server while admission is closed is rejected in `on_new_connection()` with a shutdown goodbye, the same shape as the nursery-full rejection. + +`ConnectionManager::start()` creates and configures the server object on the first call only; the client config is immutable for the client's lifetime, so a restart reuses the object and its settings. + +Each threaded role's `start()` calls `EventFlags::clear_all()` before creating its thread rather than clearing a hand-listed set of bits: a command signalled between the previous join and the restart (`cleanup()` on a stopped role) would otherwise survive into the new thread's first wait, and a bit added to the role's enum later cannot be forgotten. + +The client destructor performs steps 1 and 2 only, so a consumer that destroyed its listeners first is never called into; the roles' own destructors then join their threads as before. + +### High-performance release delivery + +`release_high_performance()` calls the listener inline. The last release can run inside `drop_connection()`, which holds `conn_ptr_mutex_` through `cleanup_connection_state()` (both the time-burst hold and the player's playback hold are released there), so the listener contract for `on_request_high_performance()` / `on_release_high_performance()` is that the body toggles the platform's networking mode and nothing else: it must not call back into the client or a role. + ### Graceful Disconnect `disconnect_and_release()` calls `conn->disconnect(reason, nullptr)` and lets the local `shared_ptr` go out of scope. @@ -471,7 +522,7 @@ Queued send workers capture a `weak_ptr` to the origin The send workers also enforce the protocol's "hello is always first" rule: a frame is dropped unless `client_hello_sent_` is set on the resolved connection, *unless* the caller passed `allow_before_hello=true`. Exactly two callers do — the `client/hello` itself (which would otherwise gate its own send and deadlock) and `goodbye` — so a stale or out-of-order frame can never precede the handshake. The `weak_ptr` guards identity; the gate guards ordering; the two are independent. -The host build does not need this scheme: `SendspinWsServer` (host) routes IXWebSocket messages by calling `find_connection_callback_` to resolve a synthetic sockfd back to the connection that `ConnectionManager` is holding. The ESP build keeps the `set_find_connection_callback()` setter as a no-op stub for symmetry; see the comment at the call site in `ConnectionManager::init_server`. +The host build does not need this scheme: `SendspinWsServer` (host) routes IXWebSocket messages by calling `find_connection_callback_` to resolve a synthetic sockfd back to the connection that `ConnectionManager` is holding. The ESP build keeps the `set_find_connection_callback()` setter as a no-op stub for symmetry; see the comment at the call site in `ConnectionManager::start`. ## Ordering Guarantees Summary diff --git a/examples/basic_client/main.cpp b/examples/basic_client/main.cpp index 9f6889de..0ae5f2c7 100644 --- a/examples/basic_client/main.cpp +++ b/examples/basic_client/main.cpp @@ -325,7 +325,7 @@ int main(int argc, char* argv[]) { // Start the server fprintf(stderr, "Starting Sendspin basic client on port %u...\n", server_port); - if (!client.start_server()) { + if (!client.start()) { fprintf(stderr, "Failed to start server\n"); return 1; } @@ -373,7 +373,7 @@ int main(int argc, char* argv[]) { #ifdef SENDSPIN_HAS_MDNS mdns.stop(); #endif - client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + client.stop(); #ifndef SENDSPIN_HAS_PORTAUDIO fprintf(stderr, "Total audio bytes received: %zu\n", null_audio_total_bytes); diff --git a/examples/tui_client/main.cpp b/examples/tui_client/main.cpp index 87fd5279..dee11746 100644 --- a/examples/tui_client/main.cpp +++ b/examples/tui_client/main.cpp @@ -763,7 +763,7 @@ int main(int argc, char* argv[]) { #endif // Start the server - if (!client.start_server()) { + if (!client.start()) { fprintf(stderr, "Failed to start server\n"); return 1; } @@ -951,7 +951,7 @@ int main(int argc, char* argv[]) { mdns_browser.stop(); mdns.stop(); #endif - client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + client.stop(); return 0; } diff --git a/include/sendspin/client.h b/include/sendspin/client.h index 5f5fdf27..e192c79f 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -68,14 +68,21 @@ class SendspinClientListener { /// @brief Called when the library needs high-performance networking (e.g., disable WiFi /// power saving) + /// + /// Toggle the platform's networking mode and return. This callback and its release can fire + /// while the client holds an internal lock (the last release runs inside the connection-loss + /// path), so the body must not call any SendspinClient or role method. virtual void on_request_high_performance() {} /// @brief Called when the library no longer needs high-performance networking + /// + /// Same contract as on_request_high_performance(): toggle the platform mode only, never call + /// back into the client. virtual void on_release_high_performance() {} }; /// @brief Platform hook for network readiness -/// Must be set before start_server() +/// Must be set before start() class SendspinNetworkProvider { public: virtual ~SendspinNetworkProvider() = default; @@ -147,8 +154,9 @@ class SendspinTimeBurst; * 2. Construct a SendspinClient with that config * 3. Add roles via add_player(), add_controller(), add_metadata(), etc. * 4. Set listeners on each role and set the network provider on the client - * 5. Call start_server() to start the WebSocket server and background tasks + * 5. Call start() to start the role threads and the WebSocket server * 6. Call loop() periodically from the platform main loop + * 7. Call stop() to goodbye every peer and tear everything down; start() again to restart * * @code * struct MyPlayerListener : PlayerRoleListener { @@ -175,11 +183,12 @@ class SendspinTimeBurst; * player.set_listener(&player_listener); * client.add_controller(); * client.set_network_provider(&network_provider); - * client.start_server(); + * client.start(); * - * while (true) { + * while (running) { * client.loop(); * } + * client.stop(); * @endcode */ class SendspinClient { @@ -201,20 +210,64 @@ class SendspinClient { // Lifecycle // ======================================== - /// @brief Starts the WebSocket server and initializes the sync task (if audio is configured) - /// @return true on success, false on failure - bool start_server(); + /// @brief Starts the role threads and arms the WebSocket server + /// + /// The server itself comes up on the first loop() tick after the network provider reports + /// ready. If a role fails to start, the roles that did start are stopped again so a corrected + /// retry begins from the stopped state. Main-loop thread only. + /// @return true if the client is running (including when it already was), false on failure + bool start(); + + /// @brief Stops the client and returns only once it is fully stopped + /// + /// Sends a client/goodbye (reason shutdown) to every peer, waits a short bound for those + /// sends to complete, then closes the server and every connection regardless, joins the role + /// threads, resets every role, and delivers the roles' clear callbacks (on_stream_end(), + /// on_image_clear(), on_metadata_clear(), ...) before returning. No-op when stopped. Calling + /// start() afterwards restarts the client; start, stop, and start again can be repeated + /// indefinitely. + /// + /// Blocking is bounded by the goodbye wait, the transports' own close, and any listener + /// callback already running on a role thread, which the join cannot interrupt. The + /// per-transport bounds are described in docs/integration-guide.md (Stopping and + /// Restarting). + /// + /// Listener callbacks fire from inside this call, after every role has been reset, so the + /// state they observe through the getters is the stopped state. One that calls start() has + /// no effect and returns false; one that calls stop(), connect_to(), or disconnect() is + /// ignored. Main-loop thread only: calling it from a role-thread callback would join the + /// calling thread. + void stop(); + + /// @brief Returns true between a successful start() and stop() + /// + /// Running means the role threads are up and the server is armed, not that the server is + /// listening yet (that waits for the network provider). Reads false for the whole duration + /// of stop(), including from the clear callbacks it fires. Safe to call from any thread. + bool is_started() const { + return this->lifecycle_.load(std::memory_order_acquire) == LifecycleState::RUNNING; + } + + /// @brief Starts the client + /// @deprecated Use start(). Kept as an alias for existing consumers; removal is planned for + /// v0.9.0. + /// @return See start(). + [[deprecated("Use start()")]] bool start_server() { + return this->start(); + } /// @brief Initiates a client connection to a Sendspin server at the given URL /// - /// Must be called from the main loop thread: it tears down and replaces connection state - /// (time filter, dispatch, client state) directly rather than deferring to loop(), so calling - /// it concurrently with loop() would race those mutations. + /// Ignored (with a warning) unless the client is running, including from a callback fired + /// inside stop(). Must be called from the main loop thread: it tears down and replaces + /// connection state (time filter, dispatch, client state) directly rather than deferring to + /// loop(), so calling it concurrently with loop() would race those mutations. /// @param url WebSocket server URL (e.g., "ws://server.local:8927/sendspin") void connect_to(const std::string& url); /// @brief Disconnects from the current server with the given reason /// + /// Ignored unless the client is running, including from a callback fired inside stop(). /// Must be called from the main loop thread: the blocking transport close runs outside the /// manager lock, so a call from another thread could race loop()'s own release of the same /// connection (two concurrent transport stops). @@ -222,10 +275,11 @@ class SendspinClient { void disconnect(SendspinGoodbyeReason reason); /// @brief Processes events, drives time sync, checks network. Call from main loop + /// A no-op while the client is stopped. void loop(); // ======================================== - // Role registration (call before start_server) + // Role registration (call before start()) // ======================================== #ifdef SENDSPIN_ENABLE_PLAYER @@ -379,7 +433,7 @@ class SendspinClient { this->listener_ = listener; } - /// @brief Sets the network provider (required before start_server()) + /// @brief Sets the network provider (required before start()) /// The provider must outlive this client void set_network_provider(SendspinNetworkProvider* provider) { this->network_provider_ = provider; @@ -405,12 +459,31 @@ class SendspinClient { void acquire_high_performance(); /// @brief Releases a ref-counted high-performance networking request + /// + /// The last release calls the listener inline, possibly under conn_ptr_mutex_ (the + /// connection-loss path); the listener contract forbids calling back into the client there. void release_high_performance(); private: /// @brief Cleans up playback state when the active streaming connection is removed void cleanup_connection_state(); + /// @brief Drains the inbox: lifecycle events, role slots, and group updates, dispatching + /// listener callbacks on the calling (main-loop) thread. Shared by loop() and stop(). + void drain_inbox(); + + /// @brief Signals the drain roles, then goodbyes and closes every transport, joining the + /// network threads. The shared first half of stop() and the destructor's teardown. + void close_transports(); + + /// @brief Asks the artwork and visualizer threads to exit without joining them, so their + /// exit overlaps the transport teardown. The player is excluded: its ring must keep a + /// consumer until the network threads are gone (see stop()). + void signal_drain_role_stops(); + + /// @brief Stops and joins every threaded role; each is a no-op if not running + void stop_role_threads(); + /// @brief Builds the formatted client hello message from config std::string build_hello_message(); @@ -502,7 +575,12 @@ class SendspinClient { // 8-bit fields bool high_performance_held_for_time_{false}; std::atomic high_performance_ref_count_{0}; - bool started_{false}; + /// Where the client is in its lifecycle. Written only by start()/stop() on the main loop; + /// atomic so is_started() can be read from any thread. STOPPING covers the whole of stop(): + /// start() is refused and stop()/connect_to()/disconnect() are ignored while it is set, so a + /// listener callback fired from inside the teardown cannot recurse into it. + enum class LifecycleState : uint8_t { STOPPED, RUNNING, STOPPING }; + std::atomic lifecycle_{LifecycleState::STOPPED}; }; } // namespace sendspin diff --git a/include/sendspin/config.h b/include/sendspin/config.h index c3675cdf..c77bf605 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -32,7 +32,7 @@ namespace sendspin { // ============================================================================ /// @brief Configuration for a SendspinClient instance -/// Filled in by the platform (e.g., ESPHome) before calling start_server() +/// Filled in by the platform (e.g., ESPHome) before calling start() struct SendspinClientConfig { /// Unique client identifier. When left empty, the library falls back to the detected local /// network interface MAC address (the same value used for device_info.mac_address). diff --git a/src/artwork_role.cpp b/src/artwork_role.cpp index a7a0465e..72e48ecf 100644 --- a/src/artwork_role.cpp +++ b/src/artwork_role.cpp @@ -101,22 +101,38 @@ bool ArtworkRole::Impl::start() { return false; } + // The flags survive a stop()/start() cycle, and a command signalled between the join and this + // start (cleanup() on a stopped role) is still set. Clear the whole group so the new thread's + // first wait() starts from a clean command state whatever bits the role defines. + this->drain_task->event_flags.clear_all(); + platform_configure_thread("SsArt", 4096, static_cast(this->config.priority), this->config.psram_stack); this->drain_task->drain_thread = std::thread(drain_thread_func, this); return true; } -void ArtworkRole::Impl::stop() const { +bool ArtworkRole::Impl::signal_stop() const { if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { - return; + return false; } // Set the flag before waking: the thread re-checks its command flags at the top of every // loop iteration, so this ordering guarantees it observes the stop as soon as the wake // pulls it out of its blocking queue receive. this->drain_task->event_flags.set(COMMAND_STOP); this->drain_task->notify_queue.wake_receiver(); + return true; +} + +void ArtworkRole::Impl::stop() const { + if (!this->signal_stop()) { + return; + } this->drain_task->drain_thread.join(); + + // Joined, so this is the queue's only consumer: discard notifications the old thread never + // took, so a restart does not decode the previous session's images. + this->drain_task->notify_queue.reset(); } void ArtworkRole::Impl::build_hello_fields(ClientHelloMessage& msg) const { diff --git a/src/artwork_role_impl.h b/src/artwork_role_impl.h index 46a6d025..1f979b0e 100644 --- a/src/artwork_role_impl.h +++ b/src/artwork_role_impl.h @@ -181,6 +181,10 @@ struct ArtworkRole::Impl { // Helpers // ======================================== + /// @brief Asks the decode thread to exit without waiting for it; stop() joins. Lets a caller + /// overlap the thread's exit with other teardown. + /// @return true if a running thread was signalled, false if none was running. + bool signal_stop() const; void stop() const; void enqueue_stream_event(ArtworkEventType event) const; // Merges a single-slot display delta into the accumulated cross-thread update. Called under diff --git a/src/client.cpp b/src/client.cpp index 0f2c27c8..e5f77d07 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -79,8 +79,24 @@ SendspinClient::SendspinClient(SendspinClientConfig config) } SendspinClient::~SendspinClient() { - // Stop background threads before tearing down connections. Every role is reset explicitly - // (not just the threaded ones): role InboxSlots release their topic-bit claims against + // Transport-only teardown: goodbye and close every peer in the same order as stop(), but + // dispatch no teardown or clear callback (nothing reaches the inbox and no drain runs). A + // role-thread callback can still run until its role is joined by the resets below, whose + // destructors run the same stop() the explicit path would, so listeners must outlive the + // client. + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + this->close_transports(); + // Every high-performance hold ends with the client. The release sites for the time hold + // (cleanup_connection_state()) and the playback hold (the player's cleanup()) do not run + // here, and nothing can acquire once the network threads are gone. + while (this->high_performance_ref_count_.load() > 0) { + this->release_high_performance(); + } + } + + // The network threads are gone (above, or never started), so the role threads are the only + // producers left; each role's destructor joins its own. Every role is reset explicitly (not + // just the threaded ones): role InboxSlots release their topic-bit claims against // event_state_'s Inbox on destruction, so all roles must be gone before the alphabetized // member order destroys event_state_. #ifdef SENDSPIN_ENABLE_PLAYER @@ -116,51 +132,152 @@ LogLevel SendspinClient::get_log_level() { // Lifecycle // ============================================================================ -bool SendspinClient::start_server() { - this->started_ = true; +bool SendspinClient::start() { + switch (this->lifecycle_.load(std::memory_order_relaxed)) { + case LifecycleState::RUNNING: + return true; + case LifecycleState::STOPPING: + SS_LOGW(TAG, "start() ignored: called from a callback while stop() is in progress"); + return false; + case LifecycleState::STOPPED: + break; + } // Load persisted state this->load_last_played_server(); + // Start the role threads. A failure part-way stops the roles that did start, so the client + // is back in the stopped state and a corrected retry begins clean. + bool roles_started = true; #ifdef SENDSPIN_ENABLE_PLAYER - if (this->player_) { - if (!this->player_->impl_->start()) { - return false; - } + if (roles_started && this->player_) { + roles_started = this->player_->impl_->start(); + } +#endif +#ifdef SENDSPIN_ENABLE_VISUALIZER + if (roles_started && this->visualizer_) { + roles_started = this->visualizer_->impl_->start(); + } +#endif +#ifdef SENDSPIN_ENABLE_ARTWORK + if (roles_started && this->artwork_) { + roles_started = this->artwork_->impl_->start(); } #endif + if (!roles_started) { + this->stop_role_threads(); + return false; + } + + // Open admission and create the WebSocket server (started by loop() once the network is + // ready). + this->connection_manager_->start(); + this->lifecycle_.store(LifecycleState::RUNNING, std::memory_order_release); + return true; +} + +void SendspinClient::stop() { + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::RUNNING) { + return; + } + // From here the client reads as stopped: is_started() is false, loop() is a no-op, and + // start()/stop()/connect_to()/disconnect() are refused, so a listener callback fired below + // cannot recurse into the teardown, restart the server, or admit a connection. + this->lifecycle_.store(LifecycleState::STOPPING, std::memory_order_release); + + // 1-2. Signal the drain roles, then goodbye and close every transport (see + // close_transports()). Nothing reaches a role or the inbox from the network after this. + this->close_transports(); + + // 3. Role threads. Each role discards its ring/queue content after its own join. + this->stop_role_threads(); + + // 4. Reset per-connection and role state exactly as a lost connection does. With every + // producer thread joined, the state this leaves behind is the state a restart begins + // from. The group slot is reset here too, so the drain below cannot repopulate + // group_state_ from a delta that arrived before the stop. + this->cleanup_connection_state(); + this->group_state_ = GroupUpdateObject{}; + this->state_ = SendspinClientState::SYNCHRONIZED; + + // 5. Deliver the clear callbacks the cleanup queued, now rather than on a loop() tick that is + // not coming. Every getter already reports the stopped state, so a callback that reads + // the client sees exactly what a caller sees once stop() returns. + this->drain_inbox(); + + this->lifecycle_.store(LifecycleState::STOPPED, std::memory_order_release); +} +void SendspinClient::close_transports() { + // 1. Ask the artwork and visualizer threads to exit now, so a slow on_image_decode() or a + // parked drain overlaps the transport teardown instead of following it. Their inbound + // channels never block a network thread (a zero-timeout queue send, a bounded ring + // acquire), so they need no consumer while the transports close. The player's ring does: + // a network thread blocked on ring space (write_audio_chunk) resolves only while the sync + // task is alive, so the player is joined by the caller after the network threads are gone. + this->signal_drain_role_stops(); + + // 2. Transports: goodbye every peer, wait up to the flush bound, then close the server and + // every connection. This joins every network thread. + this->connection_manager_->stop(SendspinGoodbyeReason::SHUTDOWN); +} + +void SendspinClient::signal_drain_role_stops() { #ifdef SENDSPIN_ENABLE_VISUALIZER if (this->visualizer_) { - if (!this->visualizer_->impl_->start()) { - return false; - } + this->visualizer_->impl_->signal_stop(); } #endif - #ifdef SENDSPIN_ENABLE_ARTWORK if (this->artwork_) { - if (!this->artwork_->impl_->start()) { - return false; - } + this->artwork_->impl_->signal_stop(); } #endif +} - // Create and configure the WebSocket server (started later when network is ready) - this->connection_manager_->init_server(this); - - return true; +void SendspinClient::stop_role_threads() { +#ifdef SENDSPIN_ENABLE_PLAYER + if (this->player_) { + this->player_->impl_->stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_VISUALIZER + if (this->visualizer_) { + this->visualizer_->impl_->stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_ARTWORK + if (this->artwork_) { + this->artwork_->impl_->stop(); + } +#endif } void SendspinClient::connect_to(const std::string& url) { + if (!this->is_started()) { + SS_LOGW(TAG, "connect_to() ignored: client is not running"); + return; + } this->connection_manager_->connect_to(url); } void SendspinClient::disconnect(SendspinGoodbyeReason reason) { + // A stopped client has nothing to disconnect, and inside stop() the manager is already + // goodbying every peer; a second pass would race the first. + if (!this->is_started()) { + SS_LOGD(TAG, "disconnect() ignored: client is not running"); + return; + } this->connection_manager_->disconnect(reason); } void SendspinClient::loop() { + // A stopped client is quiescent: no connections, no threads, and the manager loop must not + // restart the WebSocket server the moment the network reads ready. + if (!this->is_started()) { + return; + } + // Process connection lifecycle events (close, disconnect, hello, handoff, retry) this->connection_manager_->loop(); @@ -183,6 +300,10 @@ void SendspinClient::loop() { } } + this->drain_inbox(); +} + +void SendspinClient::drain_inbox() { // Process deferred events: all state mutations and user callbacks happen here, on the main // loop thread, to avoid cross-thread data races. Two poll() snapshots gate the work below: // inbox_bits (here) gates only the event-ring drain immediately following it; slot_bits @@ -253,8 +374,9 @@ void SendspinClient::loop() { // CLEARED per role is ever pending when this drain runs: cleanup() is called // only from cleanup_connection_state(), which first calls inbox.reset_events() // (wiping the whole ring) before any role re-pushes its CLEARED, and that path - // runs only under conn_ptr_mutex_ (ConnectionManager::drop_connection), so it - // cannot interleave with itself. So even a back-to-back disconnect/reconnect + // runs only on the main loop (under conn_ptr_mutex_ from + // ConnectionManager::drop_connection, or directly from stop()), so it cannot + // interleave with itself. So even a back-to-back disconnect/reconnect // coalesces to a single CLEARED -- the reset_events() ordering is what // guarantees it, not clear-callback idempotency. (Callbacks are idempotent by // contract anyway; see on_controller_state_clear() / on_metadata_clear() / @@ -391,13 +513,13 @@ void SendspinClient::loop() { } // ============================================================================ -// Role registration (call before start_server) +// Role registration (call before start()) // ============================================================================ #ifdef SENDSPIN_ENABLE_PLAYER PlayerRole& SendspinClient::add_player(PlayerRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_player() called after start_server(); role may not initialize correctly"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_player() called while started; role may not initialize correctly"); } this->player_ = std::make_unique(std::move(config), this, this->persistence_provider_); @@ -408,8 +530,8 @@ PlayerRole& SendspinClient::add_player(PlayerRoleConfig config) { #ifdef SENDSPIN_ENABLE_CONTROLLER ControllerRole& SendspinClient::add_controller() { - if (this->started_) { - SS_LOGW(TAG, "add_controller() called after start_server()"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_controller() called while started"); } this->controller_ = std::make_unique(this); this->controller_->impl_->attach_inbox(this->event_state_->inbox); @@ -419,8 +541,8 @@ ControllerRole& SendspinClient::add_controller() { #ifdef SENDSPIN_ENABLE_METADATA MetadataRole& SendspinClient::add_metadata() { - if (this->started_) { - SS_LOGW(TAG, "add_metadata() called after start_server()"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_metadata() called while started"); } this->metadata_ = std::make_unique(this); this->metadata_->impl_->attach_inbox(this->event_state_->inbox); @@ -430,8 +552,8 @@ MetadataRole& SendspinClient::add_metadata() { #ifdef SENDSPIN_ENABLE_COLOR ColorRole& SendspinClient::add_color() { - if (this->started_) { - SS_LOGW(TAG, "add_color() called after start_server()"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_color() called while started"); } this->color_ = std::make_unique(this); this->color_->impl_->attach_inbox(this->event_state_->inbox); @@ -441,8 +563,8 @@ ColorRole& SendspinClient::add_color() { #ifdef SENDSPIN_ENABLE_ARTWORK ArtworkRole& SendspinClient::add_artwork(ArtworkRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_artwork() called after start_server()"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_artwork() called while started"); } this->artwork_ = std::make_unique(std::move(config), this); this->artwork_->impl_->attach_inbox(this->event_state_->inbox); @@ -452,8 +574,8 @@ ArtworkRole& SendspinClient::add_artwork(ArtworkRoleConfig config) { #ifdef SENDSPIN_ENABLE_VISUALIZER VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_visualizer() called after start_server()"); + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_visualizer() called while started"); } this->visualizer_ = std::make_unique(std::move(config), this); this->visualizer_->impl_->attach_inbox(this->event_state_->inbox); diff --git a/src/connection_manager.cpp b/src/connection_manager.cpp index b3473e6b..7105a964 100644 --- a/src/connection_manager.cpp +++ b/src/connection_manager.cpp @@ -94,12 +94,7 @@ ConnectionManager::~ConnectionManager() { // The two mutexes guard disjoint state and are taken in separate scopes, never nested. std::vector> pending_connected; std::vector> pending_disconnects; - { - std::lock_guard lock(this->conn_mutex_); - pending_connected = std::move(this->pending_connected_events_); - pending_disconnects = std::move(this->pending_disconnect_events_); - this->has_pending_events_.store(false, std::memory_order_release); - } + this->take_pending_events(pending_connected, pending_disconnects); std::shared_ptr current; std::vector nursery; @@ -222,9 +217,21 @@ void ConnectionManager::disconnect(SendspinGoodbyeReason reason) { // Server lifecycle // ============================================================================ -void ConnectionManager::init_server(SendspinClient* client) { - this->client_ = client; +void ConnectionManager::start() { + { + std::lock_guard lock(this->conn_ptr_mutex_); + this->accepting_ = true; + } + // A restart reuses the server object: stop() only stopped it, and loop() starts it again + // once the network is ready. Retry immediately rather than honoring a backoff from before + // the stop. + this->ws_server_start_retry_time_us_ = 0; + if (this->ws_server_ != nullptr) { + return; + } + // First start: create the server object and configure it once. The config is immutable for + // the client's lifetime, so a restart reuses these values along with the object. this->ws_server_ = std::make_unique(); this->ws_server_->set_port(this->client_->config_.server_port); this->ws_server_->set_max_connections(this->client_->config_.server_max_connections); @@ -279,6 +286,82 @@ void ConnectionManager::init_server(SendspinClient* client) { }); } +void ConnectionManager::stop(SendspinGoodbyeReason reason) { + // Close admission and detach every managed connection under the lock. Nothing is sent or + // released here (see DeferredRelease): the goodbyes below run outside the lock, and a + // rejection for a peer delivered during the wait can take the lock meanwhile. + std::vector> to_goodbye; + { + std::lock_guard lock(this->conn_ptr_mutex_); + this->accepting_ = false; + if (this->current_connection_ != nullptr) { + this->current_connection_->disable_message_dispatch(); + to_goodbye.push_back(std::move(this->current_connection_)); + this->set_current_connection(nullptr); + } + for (auto& entry : this->nursery_) { + entry.conn->disable_message_dispatch(); + to_goodbye.push_back(std::move(entry.conn)); + } + this->nursery_.clear(); + this->nursery_size_.store(0, std::memory_order_release); + this->hello_retries_.clear(); + // Releases already queued (a handoff loser, a reaped entry) had their dispatch disabled + // when they were queued; the shutdown goodbye replaces whatever reason they carried. One + // queued without a reason has a transport that is already gone, and every transport's + // disconnect() completes immediately on a disconnected connection, so it needs no + // separate path. + for (auto& release : this->deferred_releases_) { + to_goodbye.push_back(std::move(release.conn)); + } + this->deferred_releases_.clear(); + this->deferred_size_.store(0, std::memory_order_release); + } + + // Goodbye every connection and wait, bounded, for the sends to complete. Every count is + // registered before the wait starts, so a completion that runs inline (host, and any + // not-connected transport) cannot satisfy the wait early. A disconnected connection completes + // immediately (see SendspinConnection::disconnect), so none needs a pre-check. + auto wait = std::make_shared(); + for (auto& conn : to_goodbye) { + wait->add_pending(); + conn->disconnect(reason, [wait] { wait->complete_one(); }); + } + // The bound scales with the goodbyes issued: on ESP they are handed to lwIP one at a time by + // the single httpd worker, so several peers need several quanta. + const uint32_t flush_bound_ms = + GOODBYE_FLUSH_TIMEOUT_MS * static_cast(to_goodbye.size()); + if (!wait->wait(flush_bound_ms)) { + SS_LOGD(TAG, "Goodbye flush bound (%u ms for %u goodbyes) elapsed; closing regardless", + static_cast(flush_bound_ms), static_cast(to_goodbye.size())); + } + + // Tear the server down regardless. This joins every network thread on host and waits for + // the httpd task on ESP, so no callback of any kind arrives after it returns. Close + // callbacks fired during it queue disconnect events under conn_mutex_, which is not held. + if (this->ws_server_ != nullptr) { + this->ws_server_->stop(); + } + + // Drop the lifecycle events those closes queued: the connections they name are gone. + std::vector> pending_connected; + std::vector> pending_disconnects; + this->take_pending_events(pending_connected, pending_disconnects); + // Locals release here, outside every lock. An outbound connection's destructor stops its + // transport synchronously; deferring that is not an option (see DeferredRelease). +} + +void ConnectionManager::take_pending_events( + std::vector>& connected, + std::vector>& disconnects) { + std::lock_guard lock(this->conn_mutex_); + connected = std::move(this->pending_connected_events_); + disconnects = std::move(this->pending_disconnect_events_); + this->pending_connected_events_.clear(); + this->pending_disconnect_events_.clear(); + this->has_pending_events_.store(false, std::memory_order_release); +} + void ConnectionManager::loop() { // Start WS server when network becomes ready. A persistent failure (e.g. the server port is // already in use) is retried with backoff instead of on every tick, which would spam the log. @@ -301,10 +384,7 @@ void ConnectionManager::loop() { // Sound because every push site sets has_pending_events_ = true under conn_mutex_ before // releasing it (see the field's doc comment in connection_manager.h). if (this->has_pending_events_.load(std::memory_order_acquire)) { - std::lock_guard lock(this->conn_mutex_); - connected_events.swap(this->pending_connected_events_); - disconnect_events.swap(this->pending_disconnect_events_); - this->has_pending_events_.store(false, std::memory_order_release); + this->take_pending_events(connected_events, disconnect_events); } // Also runs whenever the nursery is non-empty even with no swapped-out events: the @@ -585,7 +665,14 @@ void ConnectionManager::on_new_connection(std::shared_ptr= NURSERY_CAPACITY) { + if (!this->accepting_) { + // Delivered while stop() is tearing down (or before start()): the nursery is being + // emptied, so the newcomer gets a goodbye and a close instead of a slot. Same shape + // as the nursery-full rejection below. + SS_LOGD(TAG, "Not accepting connections, rejecting new connection"); + conn->disable_message_dispatch(); + this->queue_deferred_release(std::move(conn), SendspinGoodbyeReason::SHUTDOWN); + } else if (inbound_count >= NURSERY_CAPACITY) { SS_LOGW(TAG, "Nursery full of live connections, rejecting new connection"); // Never managed, but its callbacks are already wired: block dispatch so it cannot // inject messages during the goodbye window. diff --git a/src/connection_manager.h b/src/connection_manager.h index 2fc1de09..2fdd3363 100644 --- a/src/connection_manager.h +++ b/src/connection_manager.h @@ -23,6 +23,7 @@ #include "sendspin/client.h" #include +#include #include #include #include @@ -65,6 +66,55 @@ static constexpr int64_t LIVENESS_TOLERATED_MISSES = 2; /// @return Timeout in milliseconds; 0 or negative disables the check. int64_t resolve_liveness_timeout_ms(const SendspinClientConfig& config); +/// @brief Bound (milliseconds, per goodbye) on waiting for stop()'s goodbyes to be sent before +/// the transports are torn down +/// +/// stop() waits this long times the number of goodbyes it issued: on the ESP server path every +/// goodbye is queued to the single httpd worker and handed to lwIP in turn, so a fixed bound +/// would let the last of several peers lose its goodbye to the close. Per goodbye this is a few +/// scheduler quanta for the worker to dequeue the frame. The host transports send synchronously, +/// so on host the wait resolves before it starts. Send completion is best-effort (see +/// SendspinConnection::send_text_message): a session that closes first never reports, so this is +/// a cap on how long stop() blocks for its peers' sake, never a guarantee the goodbye arrived. +static constexpr uint32_t GOODBYE_FLUSH_TIMEOUT_MS = 50; + +/// @brief Counts the goodbye sends stop() is waiting on +/// +/// Shared by stop() and each connection's completion callback through a shared_ptr captured by +/// value, so a completion that runs on a transport thread after stop() has given up (an ESP httpd +/// worker draining late) touches only this record, never stop()'s stack or the manager. +struct GoodbyeWait { + /// @brief Registers one goodbye whose completion is awaited + void add_pending() { + std::lock_guard lock(this->mutex); + ++this->pending; + } + + /// @brief Records one completion; wakes wait() when none remain + void complete_one() { + { + std::lock_guard lock(this->mutex); + if (this->pending > 0) { + --this->pending; + } + } + this->cv.notify_all(); + } + + /// @brief Blocks until every registered goodbye has completed or the bound elapses + /// @param timeout_ms Maximum time to wait. + /// @return true if every goodbye completed, false if the bound elapsed first. + bool wait(uint32_t timeout_ms) { + std::unique_lock lock(this->mutex); + return this->cv.wait_for(lock, std::chrono::milliseconds(timeout_ms), + [this] { return this->pending == 0; }); + } + + std::mutex mutex; + std::condition_variable cv; + size_t pending{0}; +}; + /// @brief A connection that has not completed the hello handshake /// /// Unproven connections never occupy the current-connection slot; they wait in the bounded nursery @@ -113,21 +163,22 @@ struct HelloRetryState { * * Typical usage: * 1. Construct with a `SendspinClient*`. - * 2. Call `init_server()` once to create and configure the WebSocket server. + * 2. Call `start()` to open admission and create the WebSocket server. * 3. Call `loop()` periodically to drive connection state, process deferred events, and retry * hellos. * 4. Call `connect_to()` to initiate an outgoing client connection when needed. - * 5. Call `disconnect()` to gracefully close the active connection. + * 5. Call `disconnect()` to gracefully close the active connection, or `stop()` to tear + * every connection and the server down synchronously. * * @code * ConnectionManager manager(client); - * manager.init_server(client, use_psram, priority); + * manager.start(); * * while (running) { * manager.loop(); * } * - * manager.disconnect(SendspinGoodbyeReason::SHUTDOWN); + * manager.stop(SendspinGoodbyeReason::SHUTDOWN); * @endcode */ class ConnectionManager { @@ -156,10 +207,26 @@ class ConnectionManager { // Server lifecycle // ======================================== - /// @brief Creates the WebSocket server and configures callbacks. Call once from start_server(). - /// Server configuration is read from client->config_. - /// @param client The SendspinClient that owns this manager. - void init_server(SendspinClient* client); + /// @brief Opens admission and creates the WebSocket server on first use + /// + /// Server configuration is read from the client's config when the server object is created; + /// a restart reuses the object. loop() starts the server once the network provider reports + /// ready. Main-loop thread only. + void start(); + + /// @brief Synchronous teardown: goodbyes every managed connection, waits up to + /// GOODBYE_FLUSH_TIMEOUT_MS per goodbye for the sends to complete, then stops the WebSocket + /// server and releases every connection regardless + /// + /// Closes admission first, so a peer delivered during the wait is rejected with a goodbye. + /// Blocks on the transports' own teardown as well as the flush bound: the host server joins + /// every accepted connection thread, including a raw socket that never completed its + /// WebSocket upgrade, which can hold the join for the full WS_HANDSHAKE_TIMEOUT_SECS (3 s); + /// the ESP server waits for the httpd task to exit, and an outbound connection's transport + /// stop is synchronous (esp_websocket_client_stop() / ix::WebSocket::stop()). Client-state + /// cleanup is the caller's job: this only detaches connections. Main-loop thread only. + /// @param reason The goodbye reason sent to every connected peer. + void stop(SendspinGoodbyeReason reason); /// @brief Drives connection state: starts server when network ready, processes lifecycle /// events, retries hello, calls loop() on active connections. @@ -250,6 +317,15 @@ class ConnectionManager { /// @param conn The freshly connected connection to defer to loop(). void queue_pending_connected(std::shared_ptr conn); + /// @brief Moves both pending lifecycle event queues out under conn_mutex_ and clears + /// has_pending_events_ in the same critical section. Caller must NOT hold conn_mutex_ and + /// must let the returned connections release outside every lock (a connection destructor + /// can join its transport thread). + /// @param connected Receives pending_connected_events_. + /// @param disconnects Receives pending_disconnect_events_. + void take_pending_events(std::vector>& connected, + std::vector>& disconnects); + /// @brief Appends a connection to pending_disconnect_events_ and sets has_pending_events_ in /// the same critical section, so loop()'s lock-free gate can never miss a pushed event. /// Caller must hold conn_mutex_. @@ -345,7 +421,7 @@ class ConnectionManager { /// Socket-budget invariant: gracefully rejecting a surplus inbound peer requires the transport /// to accept NURSERY_CAPACITY + 2 sockets (1 established + the nursery + the surplus peer, /// which must be connected to receive its goodbye). The default server_max_connections - /// satisfies this; init_server warns when a configured value does not. + /// satisfies this; start() warns when a configured value does not. static constexpr size_t NURSERY_CAPACITY = 2; // Struct fields @@ -374,6 +450,10 @@ class ConnectionManager { // 8-bit fields bool has_last_played_server_{false}; + /// True between start() and stop(). Written and read only under conn_ptr_mutex_ (the read is + /// on_new_connection(), on the network thread), so a peer delivered after stop() closed + /// admission is rejected rather than admitted into a nursery stop() has already emptied. + bool accepting_{false}; // Atomic fields (lock-free hints for loop() tick gating; ground truth remains the // mutex-protected containers/pointer above -- see the "Tick cost" note on loop()) diff --git a/src/esp/ws_server.cpp b/src/esp/ws_server.cpp index 607fe8b6..65639300 100644 --- a/src/esp/ws_server.cpp +++ b/src/esp/ws_server.cpp @@ -68,8 +68,12 @@ bool SendspinWsServer::start(SendspinClient* client, bool task_stack_in_psram, config.max_open_sockets = this->max_connections_; config.open_fn = SendspinWsServer::open_callback; config.close_fn = SendspinWsServer::close_callback; + // httpd_stop() releases the global user context: with a null free function it calls plain + // free() on the pointer (esp_http_server httpd_main.c), which would free this object while + // the ConnectionManager still owns it and stop() still runs on it. A no-op free function + // keeps ownership here. config.global_user_ctx = (void*)this; - config.global_user_ctx_free_fn = nullptr; + config.global_user_ctx_free_fn = [](void* /*ctx*/) {}; // Use the configured ctrl_port, or fall back to ESP_HTTPD_DEF_CTRL_PORT + 1 to avoid // conflict with the web_server component config.ctrl_port = (this->ctrl_port_ != 0) ? this->ctrl_port_ diff --git a/src/platform/event_flags.h b/src/platform/event_flags.h index 89c6377b..5a6ad4c1 100644 --- a/src/platform/event_flags.h +++ b/src/platform/event_flags.h @@ -92,6 +92,15 @@ class EventFlags { return xEventGroupClearBits(this->handle_, bits); } + /// @brief Clears every bit, returning the group to its freshly created state + /// + /// For resetting a group whose consumer thread has been joined before a new one starts, so + /// no caller has to enumerate the bits its group defines. + /// @return Bit pattern captured before clearing. + uint32_t clear_all() { + return xEventGroupClearBits(this->handle_, USABLE_BITS); + } + /// @brief Returns the current bit pattern /// @return Current bit pattern. uint32_t get() const { @@ -110,6 +119,14 @@ class EventFlags { } private: + /// The bits a FreeRTOS event group exposes: 24 with 32-bit ticks, 8 with 16-bit ticks. The + /// upper bits are reserved by the kernel and must never be passed to the clear/set calls. +#if configUSE_16_BIT_TICKS == 1 + static constexpr uint32_t USABLE_BITS = 0x00FFU; +#else + static constexpr uint32_t USABLE_BITS = 0x00FFFFFFU; +#endif + // Pointer fields EventGroupHandle_t handle_{nullptr}; }; @@ -190,6 +207,15 @@ class EventFlags { return old; } + /// @brief Clears every bit, returning the group to its freshly created state + /// + /// For resetting a group whose consumer thread has been joined before a new one starts, so + /// no caller has to enumerate the bits its group defines. + /// @return Bit pattern captured before clearing. + uint32_t clear_all() { + return this->clear(~0U); + } + /// @brief Returns the current bit pattern /// @return Current bit pattern. uint32_t get() const { diff --git a/src/player_role.cpp b/src/player_role.cpp index 8254a2e8..8c03b718 100644 --- a/src/player_role.cpp +++ b/src/player_role.cpp @@ -175,20 +175,27 @@ void PlayerRole::Impl::attach_inbox(Inbox& inbox) { bool PlayerRole::Impl::start() { this->load_static_delay(); - if (!this->config.audio_formats.empty() && this->listener && - !this->sync_task->is_initialized()) { - if (!this->sync_task->init(this, this->client, this->config.audio_buffer_capacity)) { - SS_LOGE(TAG, "Failed to initialize sync task"); - return false; - } - if (!this->sync_task->start(this->config.psram_stack, this->config.priority)) { - SS_LOGE(TAG, "Failed to start sync task thread"); - return false; - } + if (this->config.audio_formats.empty() || !this->listener) { + return true; + } + // Init once (event flags, ring buffer); the thread is created on every start(), including a + // restart after stop(), which joined the previous one. + if (!this->sync_task->is_initialized() && + !this->sync_task->init(this, this->client, this->config.audio_buffer_capacity)) { + SS_LOGE(TAG, "Failed to initialize sync task"); + return false; + } + if (!this->sync_task->start(this->config.psram_stack, this->config.priority)) { + SS_LOGE(TAG, "Failed to start sync task thread"); + return false; } return true; } +void PlayerRole::Impl::stop() const { + this->sync_task->stop(); +} + void PlayerRole::Impl::build_hello_fields(ClientHelloMessage& msg) { if (this->config.audio_formats.empty()) { return; diff --git a/src/player_role_impl.h b/src/player_role_impl.h index 494d7053..d4df6cc6 100644 --- a/src/player_role_impl.h +++ b/src/player_role_impl.h @@ -86,6 +86,8 @@ struct PlayerRole::Impl { } void drain_events(); void cleanup(); + /// @brief Joins the sync task thread and discards its buffered audio; no-op if not started. + void stop() const; // ======================================== // Consumer-facing method implementations diff --git a/src/sync_task.cpp b/src/sync_task.cpp index 8aa088bf..c4902b73 100644 --- a/src/sync_task.cpp +++ b/src/sync_task.cpp @@ -121,10 +121,9 @@ bool SyncTask::start(bool task_stack_in_psram, unsigned priority) { return false; } - this->event_flags_.clear(EventGroupBits::TASK_RUNNING | EventGroupBits::TASK_STOPPED | - EventGroupBits::TASK_IDLE | EventGroupBits::COMMAND_STOP | - EventGroupBits::COMMAND_STREAM_END | - EventGroupBits::COMMAND_STREAM_CLEAR | EventGroupBits::COMMAND_START); + // A fresh thread starts from a clean group: no stale task state and no command signalled + // between the previous join and this start (cleanup() on a stopped task). + this->event_flags_.clear_all(); platform_configure_thread("Sendspin", SYNC_TASK_STACK_SIZE, static_cast(priority), task_stack_in_psram); @@ -802,6 +801,15 @@ void SyncTask::stop() { this->event_flags_.set(EventGroupBits::COMMAND_STOP); this->encoded_ring_buffer_->wake_receiver(); this->sync_thread_.join(); + + // A stop mid-stream leaves TASK_RUNNING set (only the idle transition clears it). The player's + // sync-idle gate reads is_running() to decide when a STREAM_END may fire, so a stopped task + // must read as idle or the stop-time on_stream_end() would wait for a thread that is gone. + this->event_flags_.clear(EventGroupBits::TASK_RUNNING); + + // The thread is joined, so this is the ring's only consumer (the single-consumer contract + // reset() requires). Discard buffered audio so a restart does not replay the old stream. + this->encoded_ring_buffer_->reset(); } // ============================================================================ @@ -817,7 +825,7 @@ void SyncTask::thread_entry(void* params) { sync_context.bytes_per_frame = sync_context.current_stream_info.frames_to_bytes(1); sync_context.decoder = std::make_unique(); - // === OUTER LOOP: persists for the lifetime of the client === + // === OUTER LOOP: persists for one started session, until stop() === while (!(this_task->event_flags_.get() & COMMAND_STOP)) { // --- IDLE STATE --- this_task->event_flags_.clear( @@ -941,6 +949,13 @@ void SyncTask::thread_entry(void* params) { // a codec header that arrived during a rapid seek (STREAM_END → STREAM_START). } + // The idle-state exits above break out while still holding the codec header they received; + // hand it back so stop()'s ring reset sees no borrowed entry. + if (sync_context.encoded_entry != nullptr) { + this_task->encoded_ring_buffer_->return_chunk(sync_context.encoded_entry); + sync_context.encoded_entry = nullptr; + } + this_task->event_flags_.set(EventGroupBits::TASK_STOPPED); } diff --git a/src/sync_task.h b/src/sync_task.h index df3de379..9a7af335 100644 --- a/src/sync_task.h +++ b/src/sync_task.h @@ -130,6 +130,11 @@ class SyncTask { /// @return true if thread started successfully, false otherwise. bool start(bool task_stack_in_psram, unsigned priority); + /// @brief Signals the task to stop, joins the thread, and discards buffered audio + /// A later start() creates a fresh thread on the same (still initialized) queues. No-op when + /// the thread is not running. Main-loop thread only: joins the sync thread. + void stop(); + /// @brief Returns true if init() has been called successfully /// @return true if the sync task has been initialized, false otherwise. bool is_initialized() const { @@ -264,9 +269,6 @@ class SyncTask { /// playtime. void process_playback_progress(SyncContext& sync_context); - /// @brief Signals the task to stop and waits for the thread to finish - void stop(); - // Struct fields EventFlags event_flags_; // Latest-wins slot that merges (sum frames, keep latest finish_timestamp) diff --git a/src/visualizer_role.cpp b/src/visualizer_role.cpp index 399e5dbc..09d9e9fb 100644 --- a/src/visualizer_role.cpp +++ b/src/visualizer_role.cpp @@ -176,22 +176,40 @@ bool VisualizerRole::Impl::start() { return false; } + // The flags survive a stop()/start() cycle, and a flush or clear signalled between the join + // and this start (cleanup() on a stopped role) is still set. Clear the whole group so the new + // thread starts from a clean command state whatever bits the role defines (stop() already + // emptied the ring). + this->drain_task->event_flags.clear_all(); + platform_configure_thread("SsVis", 4096, static_cast(this->config.priority), this->config.psram_stack); this->drain_task->drain_thread = std::thread(drain_thread_func, this); return true; } -void VisualizerRole::Impl::stop() const { +bool VisualizerRole::Impl::signal_stop() const { if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { - return; + return false; } // Set the flag before waking: the thread re-checks its command flags at the top of every // loop iteration, so this ordering guarantees it observes the stop no matter which wait // it was parked in (display-time flags wait or ring buffer receive). this->drain_task->event_flags.set(COMMAND_STOP); this->drain_task->ring_buffer.wake_receiver(); + return true; +} + +void VisualizerRole::Impl::stop() const { + if (!this->signal_stop()) { + return; + } this->drain_task->drain_thread.join(); + + // Joined, so this is the ring's only consumer (the single-consumer contract the ring + // requires): discard entries the old thread never took, so a restart does not deliver the + // previous session's frames against the new session's format. + this->flush_ring_buffer(); } void VisualizerRole::Impl::build_hello_fields(ClientHelloMessage& msg) { diff --git a/src/visualizer_role_impl.h b/src/visualizer_role_impl.h index 56cad0c0..fc8d6af4 100644 --- a/src/visualizer_role_impl.h +++ b/src/visualizer_role_impl.h @@ -111,6 +111,10 @@ struct VisualizerRole::Impl { // Internal helpers // ======================================== + /// @brief Asks the drain thread to exit without waiting for it; stop() joins. Lets a caller + /// overlap the thread's exit with other teardown. + /// @return true if a running thread was signalled, false if none was running. + bool signal_stop() const; void stop() const; void flush_ring_buffer() const; void signal_clear_marker() const; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a429a078..cd67a861 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -33,8 +33,13 @@ add_executable(sendspin_tests test_visualizer_role.cpp test_artwork_role.cpp test_client_teardown.cpp + test_client_lifecycle.cpp ) +# test_client_lifecycle.cpp checks what a role's stop() left in its ring, which nothing on the +# public surface reports; it reads the private members directly instead of adding a seam. +set_source_files_properties(test_client_lifecycle.cpp PROPERTIES COMPILE_OPTIONS -fno-access-control) + # Reach the library's private headers (protocol_messages.h, time_filter.h, ...). # The public include/ dir and ArduinoJson propagate transitively from `sendspin`. # Use CMAKE_CURRENT_SOURCE_DIR (not CMAKE_SOURCE_DIR) so the path stays correct even diff --git a/tests/test_artwork_role.cpp b/tests/test_artwork_role.cpp index 43b679f9..545fae3e 100644 --- a/tests/test_artwork_role.cpp +++ b/tests/test_artwork_role.cpp @@ -702,6 +702,81 @@ TEST(ArtworkFrameDoneGate, RestartKeepsPresentedGate) { EXPECT_EQ(listener.decode_marker_at(1), 'B'); } +// ============================================================================ +// Impl stop()/start(): the decode thread is joined and restarted between sessions +// ============================================================================ + +namespace { + +// A RecordingListener whose on_image_decode() parks until release(), so a test can hold the +// decode thread inside a callback while it queues more work behind it. +class BlockingListener : public RecordingListener { +public: + void on_image_decode(uint8_t slot, const uint8_t* data, size_t length, + SendspinImageFormat format) override { + RecordingListener::on_image_decode(slot, data, length, format); + std::unique_lock lock(this->gate_mutex_); + this->gate_cv_.wait(lock, [this] { return this->released_; }); + } + + void release() { + { + std::lock_guard lock(this->gate_mutex_); + this->released_ = true; + } + this->gate_cv_.notify_all(); + } + +private: + std::mutex gate_mutex_; + std::condition_variable gate_cv_; + bool released_{false}; +}; + +// Two ungated slots, so a frame on each is decoded without an ack. +ArtworkRoleConfig make_two_ungated_slot_config() { + ArtworkRoleConfig config; + config.preferred_formats.push_back( + {SendspinImageSource::ALBUM, SendspinImageFormat::JPEG, 100, 100, false}); + config.preferred_formats.push_back( + {SendspinImageSource::ARTIST, SendspinImageFormat::JPEG, 100, 100, false}); + return config; +} + +} // namespace + +// stop() joins the decode thread and discards the notifications it never took, and start() +// clears the stop command, so a restarted role decodes fresh frames without replaying the +// previous session's. The thread is held inside frame A's decode while frame B is queued behind +// it and the stop is signalled; on release it exits at its command check without taking B. The +// stream is deliberately not restarted after start(): a stream restart bumps the epoch that +// would make a replayed B stale on its own, and this test is about the queue reset. +TEST(ArtworkRestart, StopDiscardsQueuedFramesAndStartDecodesNewOnes) { + BlockingListener listener; + auto impl = make_impl(make_two_ungated_slot_config()); + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'A'); + listener.wait_until([&] { return listener.decodes.size() >= 1; }); // Thread parked in A + send_frame(*impl, 1, 'B'); // Queued behind A + + ASSERT_TRUE(impl->signal_stop()); + listener.release(); + impl->stop(); + EXPECT_EQ(listener.decode_count(), 1U); + + ASSERT_TRUE(impl->start()); + // B was discarded with the old session, not replayed by the new thread. + EXPECT_TRUE(listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + // The new thread decodes: the stop command did not survive the restart. + send_frame(*impl, 1, 'C'); + listener.wait_until([&] { return listener.decodes.size() >= 2; }); + EXPECT_EQ(listener.decode_marker_at(1), 'C'); +} + // ============================================================================ // Reentrant frame_done() from inside on_image_display() // ============================================================================ diff --git a/tests/test_client_lifecycle.cpp b/tests/test_client_lifecycle.cpp new file mode 100644 index 00000000..f7bdbe9c --- /dev/null +++ b/tests/test_client_lifecycle.cpp @@ -0,0 +1,563 @@ +// Copyright 2026 Sendspin Contributors +// +// 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. + +/// @file test_client_lifecycle.cpp +/// @brief start() / stop() / restart of a SendspinClient: peers are goodbyed, role state is +/// reset and its clear callbacks delivered before stop() returns, a restarted client is live +/// again, and a callback fired from inside stop() cannot recurse into the lifecycle. +/// +/// The client is driven on loopback ports like test_connection_lifecycle.cpp: an IXWebSocket +/// endpoint plays the Sendspin server and the test thread pumps client.loop(). + +#include "connection_manager.h" // GoodbyeWait, GOODBYE_FLUSH_TIMEOUT_MS +#include "platform/time.h" +#include "protocol_messages.h" // SENDSPIN_BINARY_VISUALIZER_LOUDNESS +#include "sendspin/client.h" +#include "sendspin/config.h" +#include "sendspin/metadata_role.h" +#include "sendspin/player_role.h" +#include "sendspin/visualizer_role.h" +#include "test_support.h" +#include "visualizer_role_impl.h" // Ring state after stop(); private access, see tests/CMakeLists.txt + +#include + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience +using namespace sendspin::test; // NOLINT(google-build-using-namespace): shared loopback scaffolding + +namespace { + +// Distinct ports per test so a lingering socket from one scenario cannot bleed into the next +// (and into test_connection_lifecycle.cpp, which uses 18941-18985). +constexpr uint16_t RESTART_TEST_PORT = 18991; +constexpr uint16_t NURSERY_GOODBYE_TEST_PORT = 18992; +constexpr uint16_t STREAM_TEST_PORT = 18993; +constexpr uint16_t CALLBACK_TEST_PORT = 18994; +constexpr uint16_t DESTRUCTOR_TEST_PORT = 18995; +constexpr uint16_t ROLLBACK_TEST_PORT = 18996; +constexpr uint16_t HIGH_PERF_TEST_PORT = 18997; +constexpr uint16_t VISUALIZER_TEST_PORT = 18998; +constexpr uint16_t DESTRUCTOR_HIGH_PERF_TEST_PORT = 18999; + +/// Reports whether anything is listening on the loopback port. +bool port_accepts(uint16_t port) { + int fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + return false; + } + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(port); + const bool connected = ::connect(fd, reinterpret_cast(&addr), sizeof(addr)) == 0; + ::close(fd); + return connected; +} + +/// Counts the player lifecycle callbacks and audio writes; the write itself is a sink. +class CountingPlayerListener : public PlayerRoleListener { +public: + size_t on_audio_write(uint8_t* /*data*/, size_t length, uint32_t /*timeout_ms*/) override { + this->audio_writes.fetch_add(1); + return length; + } + void on_stream_start() override { + ++this->stream_starts; + } + void on_stream_end() override { + ++this->stream_ends; + } + + std::atomic audio_writes{0}; + int stream_starts{0}; + int stream_ends{0}; +}; + +/// Records on_metadata_clear() and, from inside it, tries to drive the lifecycle re-entrantly. +class ReentrantMetadataListener : public MetadataRoleListener { +public: + explicit ReentrantMetadataListener(SendspinClient& client) : client_(client) {} + + void on_metadata_clear() override { + ++this->clears; + this->started_during_clear = this->client_.is_started(); + this->group_had_state_during_clear = + this->client_.get_group_state().playback_state.has_value(); + this->start_result_during_clear = this->client_.start(); + this->client_.stop(); // Must be ignored, not recurse + this->client_.disconnect(SendspinGoodbyeReason::SHUTDOWN); // Must be ignored + this->client_.connect_to("ws://127.0.0.1:1/sendspin"); // Must be ignored + } + + int clears{0}; + bool started_during_clear{true}; + bool group_had_state_during_clear{true}; + bool start_result_during_clear{true}; + +private: + SendspinClient& client_; +}; + +/// A metadata listener that must never be called; every callback aborts the test. +class ForbiddenMetadataListener : public MetadataRoleListener { +public: + void on_metadata(const ServerMetadataStateObject& /*metadata*/) override { + ADD_FAILURE() << "on_metadata() fired on a listener the consumer already released"; + } + void on_metadata_clear() override { + ADD_FAILURE() << "on_metadata_clear() fired on a listener the consumer already released"; + } +}; + +std::string stream_start_pcm_json() { + return R"({"type":"stream/start","payload":{"player":{"codec":"pcm","sample_rate":48000,)" + R"("channels":2,"bit_depth":16}}})"; +} + +PlayerRoleConfig make_player_config() { + PlayerRoleConfig player_cfg; + player_cfg.audio_formats.push_back({SendspinCodecFormat::PCM, 2, 48000, 16}); + player_cfg.audio_buffer_capacity = 64 * 1024; + return player_cfg; +} + +// Pumps until the peer has written at least `target` audio callbacks, feeding 20 ms PCM chunks +// stamped a little ahead of now so the sync task has something to schedule. +void stream_audio_until(SendspinClient& client, FakeServer& server, CountingPlayerListener& listener, + size_t target) { + constexpr size_t PCM_20MS_BYTES = 48000 / 50 * 2 * 2; + int64_t next_ts = platform_time_us() + 50 * 1000; + pump_until(client, [&] { + if (listener.audio_writes.load() >= target) { + return true; + } + server.send_audio(next_ts, PCM_20MS_BYTES); + next_ts += 20 * 1000; + std::this_thread::sleep_for(std::chrono::milliseconds(20)); // real-time pacing + return false; + }); +} + +// ============================================================================ +// GoodbyeWait: the bound stop() relies on +// ============================================================================ + +// A goodbye whose completion never arrives (an ESP session that closes before its worker runs +// reports nothing) must not hold stop() open: wait() returns false once the bound elapses. +// Deleting the bound turns this into a hang the suite watchdog reports. +TEST(GoodbyeWait, BoundElapsesWhenACompletionNeverArrives) { + GoodbyeWait wait; + wait.add_pending(); + EXPECT_FALSE(wait.wait(GOODBYE_FLUSH_TIMEOUT_MS)); +} + +// Control: with every registered goodbye completed (from another thread, as a transport worker +// would) wait() reports success, and with nothing registered it never blocks. +TEST(GoodbyeWait, CompletionsSatisfyTheWait) { + GoodbyeWait idle; + EXPECT_TRUE(idle.wait(GOODBYE_FLUSH_TIMEOUT_MS)); + + GoodbyeWait wait; + wait.add_pending(); + wait.add_pending(); + std::thread worker([&] { + wait.complete_one(); + wait.complete_one(); + }); + // No bound: a lost completion hangs here and the watchdog reports it, rather than the + // elapsed time deciding the verdict. + EXPECT_TRUE(wait.wait(UINT32_MAX)); + worker.join(); +} + +// ============================================================================ +// SendspinClient lifecycle +// ============================================================================ + +// start -> stop -> start, twice over: every stop goodbyes and closes the established peer, resets +// the group state, and leaves nothing listening; every restart accepts a new peer and completes +// its handshake. Also pins start() as idempotent while running and stop() as a no-op when +// stopped. +TEST(ClientLifecycle, RestartYieldsALiveClient) { + TestNetworkProvider network; + SendspinClient client(make_config(RESTART_TEST_PORT)); + client.set_network_provider(&network); + + EXPECT_FALSE(client.is_started()); + client.stop(); // No-op when stopped + EXPECT_FALSE(client.is_started()); + + for (int cycle = 0; cycle < 3; ++cycle) { + ASSERT_TRUE(client.start()); + EXPECT_TRUE(client.start()); // Already running: reports true, starts nothing twice + EXPECT_TRUE(client.is_started()); + + const std::string server_id = "server-" + std::to_string(cycle); + FakeServer server(server_url(RESTART_TEST_PORT), server_id); + pump_until(client, [&] { return client.is_connected(); }); + auto info = client.get_server_information(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->server_id, server_id); + + // Some group state for stop() to reset. + server.send_text(R"({"type":"group/update","payload":{"playback_state":"playing"}})"); + pump_until(client, [&] { + return client.get_group_state().playback_state.has_value(); + }); + + client.stop(); + + EXPECT_FALSE(client.is_started()); + EXPECT_FALSE(client.is_connected()); + EXPECT_FALSE(client.get_server_information().has_value()); + EXPECT_FALSE(client.get_group_state().playback_state.has_value()); + // The peer received its goodbye and the close, in that order. + wait_until([&] { return server.closed(); }); + EXPECT_TRUE(server.got_goodbye()); + + // Stopped means quiescent: pumping loop() must not bring the server back up. + pump_for(client, 100); + EXPECT_FALSE(port_accepts(RESTART_TEST_PORT)); + } +} + +// A peer still in the nursery (it upgraded but never answered the hello) gets the same goodbye and +// close as the established one, so no peer is left to discover the shutdown by timeout. +TEST(ClientLifecycle, StopGoodbyesNurseryPeersToo) { + TestNetworkProvider network; + SendspinClient client(make_config(NURSERY_GOODBYE_TEST_PORT)); + client.set_network_provider(&network); + ASSERT_TRUE(client.start()); + + FakeServer established(server_url(NURSERY_GOODBYE_TEST_PORT), "server-established"); + pump_until(client, [&] { return client.is_connected(); }); + + FakeServer mute(server_url(NURSERY_GOODBYE_TEST_PORT), "server-mute", + FakeServerOptions{.answer_hello = false}); + pump_until(client, [&] { return mute.got_client_hello(); }); + + client.stop(); + + wait_until([&] { return established.closed() && mute.closed(); }); + EXPECT_TRUE(established.got_goodbye()); + EXPECT_TRUE(mute.got_goodbye()); +} + +// With a stream playing, stop() ends it (on_stream_end() fires before stop() returns, paired with +// the earlier on_stream_start()) and a restarted client plays a new stream: audio reaches the +// listener again, which needs the sync task thread to have been re-created, not just the server. +TEST(ClientLifecycle, StopEndsTheStreamAndRestartPlaysAgain) { + TestNetworkProvider network; + CountingPlayerListener listener; + auto config = make_config(STREAM_TEST_PORT); + config.time_burst_interval_ms = 100; // Sync promptly after each (re)connect + SendspinClient client(std::move(config)); + client.set_network_provider(&network); + client.add_player(make_player_config()).set_listener(&listener); + + for (int cycle = 0; cycle < 2; ++cycle) { + ASSERT_TRUE(client.start()); + FakeServer server(server_url(STREAM_TEST_PORT), "server-" + std::to_string(cycle), + FakeServerOptions{.answer_time = true}); + pump_until(client, [&] { return client.is_connected(); }); + + server.send_text(stream_start_pcm_json()); + pump_until(client, [&] { return listener.stream_starts == cycle + 1; }); + EXPECT_EQ(listener.stream_ends, cycle); + + // Audio flowing proves the sync task thread is alive in this cycle. + const size_t writes_before = listener.audio_writes.load(); + stream_audio_until(client, server, listener, writes_before + 1); + + client.stop(); + + // The clear callback was delivered inside stop(), not left for a loop() tick. + EXPECT_EQ(listener.stream_ends, cycle + 1); + EXPECT_EQ(listener.stream_starts, cycle + 1); + wait_until([&] { return server.closed(); }); + EXPECT_TRUE(server.got_goodbye()); + } +} + +// A listener callback fired from inside stop() cannot re-enter the lifecycle: start() reports +// failure and starts nothing, stop()/disconnect()/connect_to() are ignored rather than recursing, +// and the client (its started flag and its group state) already reads as stopped. Afterwards the +// client restarts normally. +TEST(ClientLifecycle, CallbackDuringStopCannotRecurse) { + TestNetworkProvider network; + SendspinClient client(make_config(CALLBACK_TEST_PORT)); + client.set_network_provider(&network); + ReentrantMetadataListener listener(client); + client.add_metadata().set_listener(&listener); + ASSERT_TRUE(client.start()); + + { + FakeServer server(server_url(CALLBACK_TEST_PORT), "server-a"); + pump_until(client, [&] { return client.is_connected(); }); + // Group state the callback must already see reset. + server.send_text(R"({"type":"group/update","payload":{"playback_state":"playing"}})"); + pump_until(client, [&] { return client.get_group_state().playback_state.has_value(); }); + + client.stop(); + + EXPECT_EQ(listener.clears, 1); + EXPECT_FALSE(listener.started_during_clear); + EXPECT_FALSE(listener.group_had_state_during_clear); + EXPECT_FALSE(listener.start_result_during_clear); + EXPECT_FALSE(client.is_started()); + wait_until([&] { return server.closed(); }); + } + + // The refused start() inside the callback left the client stopped; a real start() works. + ASSERT_TRUE(client.start()); + FakeServer server(server_url(CALLBACK_TEST_PORT), "server-b"); + pump_until(client, [&] { return client.is_connected(); }); + client.stop(); + EXPECT_EQ(listener.clears, 2); +} + +// Destroying a running client goodbyes its peer like stop() does, but dispatches no clear +// callback: the listener outlives the client, as the role contract requires, and fails the test +// if the destructor calls into it. +TEST(ClientLifecycle, DestructorGoodbyesPeersWithoutCallbacks) { + TestNetworkProvider network; + FakeServer* server = nullptr; + ForbiddenMetadataListener listener; + { + SendspinClient client(make_config(DESTRUCTOR_TEST_PORT)); + client.set_network_provider(&network); + client.add_metadata().set_listener(&listener); + ASSERT_TRUE(client.start()); + + server = new FakeServer(server_url(DESTRUCTOR_TEST_PORT), "server-a"); + pump_until(client, [&] { return client.is_connected(); }); + // Client destroyed here while established. + } + + wait_until([&] { return server->closed(); }); + EXPECT_TRUE(server->got_goodbye()); + delete server; +} + +// A role that fails to start part-way through start() rolls the roles before it back: here the +// player comes up and the visualizer (a ring too small to create) refuses, so start() reports +// failure and the client stays stopped. Replacing the broken role and starting again succeeds, +// which needs the first attempt to have joined the player's sync task: SyncTask::start() refuses +// a thread that is still running, so a rollback that skipped the join fails the retry too. +TEST(ClientLifecycle, FailedRoleStartRollsBackAndRetryStartsClean) { + TestNetworkProvider network; + CountingPlayerListener listener; + SendspinClient client(make_config(ROLLBACK_TEST_PORT)); + client.set_network_provider(&network); + client.add_player(make_player_config()).set_listener(&listener); + + VisualizerRoleConfig broken; + broken.support.types = {VisualizerDataType::LOUDNESS}; + broken.support.buffer_capacity = 0; // Below the ring's minimum: start() fails + broken.support.rate_max = 30; + client.add_visualizer(std::move(broken)); + + EXPECT_FALSE(client.start()); + EXPECT_FALSE(client.is_started()); + EXPECT_FALSE(client.start()); // Still broken, still refused, still not stuck half-started + + VisualizerRoleConfig working; + working.support.types = {VisualizerDataType::LOUDNESS}; + working.support.buffer_capacity = 4096; + working.support.rate_max = 30; + client.add_visualizer(std::move(working)); + + ASSERT_TRUE(client.start()); + FakeServer server(server_url(ROLLBACK_TEST_PORT), "server-a", + FakeServerOptions{.answer_time = true}); + pump_until(client, [&] { return client.is_connected(); }); + server.send_text(stream_start_pcm_json()); + pump_until(client, [&] { return listener.stream_starts == 1; }); + stream_audio_until(client, server, listener, 1); // The rolled-back player plays again + client.stop(); + EXPECT_EQ(listener.stream_ends, 1); +} + +/// Counts loudness deliveries; they fire on the visualizer drain thread. +class CountingVisualizerListener : public VisualizerRoleListener { +public: + void on_loudness(int64_t /*client_timestamp*/, uint16_t /*loudness*/) override { + this->loudness.fetch_add(1); + } + + std::atomic loudness{0}; +}; + +std::string stream_start_visualizer_json() { + return R"({"type":"stream/start","payload":{"visualizer":{"types":["loudness"],"rate_max":30}}})"; +} + +VisualizerRoleConfig make_visualizer_config() { + VisualizerRoleConfig config; + config.support.types = {VisualizerDataType::LOUDNESS}; + config.support.buffer_capacity = 4096; + config.support.rate_max = 30; + return config; +} + +// Waits for a fresh peer that answers time messages to be established and synced: the drain +// thread delivers nothing until the client is time synced. +void pump_until_synced(SendspinClient& client) { + pump_until(client, [&] { return client.is_connected() && client.is_time_synced(); }); +} + +// Pumps until pred() holds, sending one loudness frame per iteration stamped `lead_us` ahead of +// the current time (the drain thread drops a frame whose display time is well past). A frame +// can be lost to the ring's documented wake race right after a stream/start (the drain thread +// may take the clear marker as a stray entry and then discard up to a marker that is gone), +// which production shrugs off because the next frame follows; so does this. +void send_loudness_until(SendspinClient& client, FakeServer& server, int64_t lead_us, + const std::function& pred) { + pump_until(client, [&] { + if (pred()) { + return true; + } + server.send_binary(SENDSPIN_BINARY_VISUALIZER_LOUDNESS, platform_time_us() + lead_us, + std::string("\x00\x10", 2)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + return false; + }); +} + +// stop() joins the visualizer drain thread and flushes the frames it had buffered, and start() +// clears the stop command, so a restart begins with an empty ring and a thread that delivers. +// The old frames are stamped far into the future, so the first session's thread parks on the +// first one with the rest buffered behind it when stop() runs; the ring is read directly after +// the stop because the restarted thread would silently drop leftovers before the new peer is +// time synced, and the new session's stream/start would discard them at its clear marker. +TEST(ClientLifecycle, StopFlushesBufferedVisualizerFramesAndRestartDelivers) { + constexpr int64_t OLD_FRAME_LEAD_US = 5 * 1000 * 1000; + + TestNetworkProvider network; + CountingVisualizerListener listener; + auto config = make_config(VISUALIZER_TEST_PORT); + config.time_burst_interval_ms = 100; // Sync promptly after each (re)connect + SendspinClient client(std::move(config)); + client.set_network_provider(&network); + client.add_visualizer(make_visualizer_config()).set_listener(&listener); + + ASSERT_TRUE(client.start()); + { + FakeServer server(server_url(VISUALIZER_TEST_PORT), "server-a", + FakeServerOptions{.answer_time = true}); + pump_until_synced(client); + server.send_text(stream_start_visualizer_json()); + // The thread holds the first frame while it waits for its display time; the ones behind + // it are the ring content stop() must discard. + auto& ring = client.visualizer()->impl_->drain_task->ring_buffer; + send_loudness_until(client, server, OLD_FRAME_LEAD_US, + [&] { return ring.items_waiting() >= 2; }); + client.stop(); + EXPECT_TRUE(ring.is_empty()); + wait_until([&] { return server.closed(); }); + } + EXPECT_EQ(listener.loudness.load(), 0U); + + ASSERT_TRUE(client.start()); + FakeServer server(server_url(VISUALIZER_TEST_PORT), "server-b", + FakeServerOptions{.answer_time = true}); + pump_until_synced(client); + server.send_text(stream_start_visualizer_json()); + send_loudness_until(client, server, 0, [&] { return listener.loudness.load() >= 1; }); + client.stop(); +} + +/// Counts high-performance requests and releases without touching the client, as the listener +/// contract requires. +class CountingClientListener : public SendspinClientListener { +public: + void on_request_high_performance() override { + ++this->requests; + } + void on_release_high_performance() override { + ++this->releases; + } + + int requests{0}; + int releases{0}; +}; + +// The high-performance hold taken for a time burst is released inside the connection-loss path +// and again by stop(); request and release stay paired across a peer loss, a reconnect, and the +// stop. +TEST(ClientLifecycle, HighPerformanceRequestAndReleaseStayPaired) { + TestNetworkProvider network; + auto config = make_config(HIGH_PERF_TEST_PORT); + config.time_burst_interval_ms = 50; + SendspinClient client(std::move(config)); + client.set_network_provider(&network); + CountingClientListener listener; + client.set_listener(&listener); + ASSERT_TRUE(client.start()); + + auto server = std::make_unique(server_url(HIGH_PERF_TEST_PORT), "server-a"); + pump_until(client, [&] { return client.is_connected(); }); + // The default FakeServer never answers client/time, so the burst stays open and the hold + // stays held until the connection is lost. + pump_until(client, [&] { return listener.requests == 1; }); + EXPECT_EQ(listener.releases, 0); + + server.reset(); // Peer goes away mid-burst: drop_connection releases the hold + pump_until(client, [&] { return listener.releases == 1; }); + EXPECT_FALSE(client.is_connected()); + + FakeServer again(server_url(HIGH_PERF_TEST_PORT), "server-b"); + pump_until(client, [&] { return client.is_connected() && listener.requests == 2; }); + client.stop(); + EXPECT_EQ(listener.releases, 2); +} + +// Destroying a running client ends the hold too: the release sites stop() reaches through the +// connection cleanup do not run in the destructor, so it has to release on its own or the +// platform is left in high-performance mode after the client is gone. +TEST(ClientLifecycle, DestructorReleasesHighPerformanceHold) { + TestNetworkProvider network; + CountingClientListener listener; + { + auto config = make_config(DESTRUCTOR_HIGH_PERF_TEST_PORT); + config.time_burst_interval_ms = 50; + SendspinClient client(std::move(config)); + client.set_network_provider(&network); + client.set_listener(&listener); + ASSERT_TRUE(client.start()); + + FakeServer server(server_url(DESTRUCTOR_HIGH_PERF_TEST_PORT), "server-a"); + pump_until(client, [&] { return client.is_connected() && listener.requests == 1; }); + EXPECT_EQ(listener.releases, 0); + // Client destroyed here mid-burst, with the hold open. + } + EXPECT_EQ(listener.releases, 1); +} + +} // namespace diff --git a/tests/test_client_teardown.cpp b/tests/test_client_teardown.cpp index 0dbd4c29..3e5bdb72 100644 --- a/tests/test_client_teardown.cpp +++ b/tests/test_client_teardown.cpp @@ -81,7 +81,7 @@ TEST(ClientTeardown, JoinsEveryThreadedRoleOnDestruction) { vis_cfg.support.rate_max = 30; client->add_visualizer(std::move(vis_cfg)); - ASSERT_TRUE(client->start_server()); + ASSERT_TRUE(client->start()); if (run > 0) { // Run 0 tears down mid-startup; later runs tear down threads parked in receives. diff --git a/tests/test_connection_lifecycle.cpp b/tests/test_connection_lifecycle.cpp index f44abd36..44811c35 100644 --- a/tests/test_connection_lifecycle.cpp +++ b/tests/test_connection_lifecycle.cpp @@ -22,6 +22,7 @@ #include "connection_manager.h" // fnv1_hash, resolve_liveness_timeout_ms #include "sendspin/client.h" #include "sendspin/config.h" +#include "test_support.h" #include #include #include @@ -41,7 +42,8 @@ #include #include -using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience +using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience +using namespace sendspin::test; // NOLINT(google-build-using-namespace): shared loopback scaffolding namespace { @@ -60,35 +62,6 @@ constexpr uint16_t LIVENESS_TEST_PORT = 18983; constexpr uint16_t LIVENESS_CONTROL_PORT = 18984; constexpr uint16_t LIVENESS_DISABLED_PORT = 18985; -std::string server_url(uint16_t port) { - return "ws://127.0.0.1:" + std::to_string(port) + "/sendspin"; -} - -std::string server_hello_json(const std::string& server_id, const std::string& reason) { - return std::string(R"({"type":"server/hello","payload":{"server_id":")") + server_id + - R"(","name":"Fake Server","version":1,"active_roles":["player"],)" + - R"("connection_reason":")" + reason + R"("}})"; -} - -// Any complete inbound frame counts for liveness, so the reply's timestamps need not be real. -constexpr const char* SERVER_TIME_JSON = - R"({"type":"server/time","payload":{"client_transmitted":0,"server_received":1000,"server_transmitted":1001}})"; - -SendspinClientConfig make_config(uint16_t port) { - SendspinClientConfig config; - config.client_id = "lifecycle-test-client"; - config.name = "Lifecycle Test Client"; - config.server_port = port; - return config; -} - -class TestNetworkProvider : public SendspinNetworkProvider { -public: - bool is_network_ready() override { - return true; - } -}; - class TestPersistenceProvider : public SendspinPersistenceProvider { public: explicit TestPersistenceProvider(uint32_t hash) : hash_(hash) {} @@ -101,27 +74,6 @@ class TestPersistenceProvider : public SendspinPersistenceProvider { uint32_t hash_; }; -// Pumps client.loop() until pred() is true. No timeout: a regression hangs here and the CTest -// TIMEOUT reports it. -void pump_until(SendspinClient& client, const std::function& pred) { - for (;;) { - client.loop(); - if (pred()) { - return; - } - std::this_thread::sleep_for(std::chrono::milliseconds(2)); - } -} - -// Pumps client.loop() for a fixed window. Only for spacing events or "must not happen" checks: -// a window that is too short can miss a regression, never fail a correct run. -void pump_for(SendspinClient& client, int duration_ms) { - const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(duration_ms); - while (std::chrono::steady_clock::now() < deadline) { - client.loop(); - std::this_thread::sleep_for(std::chrono::milliseconds(2)); - } -} int connect_loopback(uint16_t port) { int fd = ::socket(AF_INET, SOCK_STREAM, 0); @@ -155,93 +107,6 @@ bool socket_closed(int fd) { } } -/// Behavior knobs for FakeServer. -struct FakeServerOptions { - bool hello_on_open{false}; ///< Send server/hello immediately on Open, before any - ///< client/hello arrives (a nonconforming peer) - bool answer_hello{true}; ///< Reply to client/hello with server/hello (false: mute peer - ///< that upgrades and then never establishes) - bool answer_time{false}; ///< Reply to client/time with server/time (a live server); false - ///< models a peer whose socket went silent after establishing -}; - -/// A minimal Sendspin "server": an IXWebSocket client that connects to the SendspinClient's WS -/// server (the server-initiated discovery direction) and answers client/hello with server/hello, -/// per the given options. -class FakeServer { -public: - FakeServer(const std::string& url, std::string server_id, FakeServerOptions options = {}) - : server_id_(std::move(server_id)) { - this->ws_.setUrl(url); - this->ws_.disableAutomaticReconnection(); - this->ws_.setOnMessageCallback([this, options](const ix::WebSocketMessagePtr& msg) { - if (msg->type == ix::WebSocketMessageType::Open) { - if (options.hello_on_open) { - this->ws_.send(server_hello_json(this->server_id_, "discovery")); - } - } else if (msg->type == ix::WebSocketMessageType::Message && - msg->str.find("client/hello") != std::string::npos) { - this->got_client_hello_.store(true); - if (options.answer_hello) { - this->ws_.send(server_hello_json(this->server_id_, "discovery")); - } - } else if (msg->type == ix::WebSocketMessageType::Message && - msg->str.find("client/goodbye") != std::string::npos) { - { - std::lock_guard lock(this->goodbye_mutex_); - this->goodbye_message_ = msg->str; - } - this->got_goodbye_.store(true); - } else if (msg->type == ix::WebSocketMessageType::Message && - msg->str.find("client/time") != std::string::npos) { - this->got_client_time_.store(true); - if (options.answer_time) { - this->ws_.send(SERVER_TIME_JSON); - } - } else if (msg->type == ix::WebSocketMessageType::Close || - msg->type == ix::WebSocketMessageType::Error) { - this->closed_.store(true); - } - }); - this->ws_.start(); - } - - ~FakeServer() { - this->ws_.stop(); - } - - bool closed() const { - return this->closed_.load(); - } - - bool got_client_hello() const { - return this->got_client_hello_.load(); - } - - bool got_goodbye() const { - return this->got_goodbye_.load(); - } - - std::string goodbye_message() const { - std::lock_guard lock(this->goodbye_mutex_); - return this->goodbye_message_; - } - - bool got_client_time() const { - return this->got_client_time_.load(); - } - -private: - ix::WebSocket ws_; - std::string server_id_; - mutable std::mutex goodbye_mutex_; - std::string goodbye_message_; - std::atomic closed_{false}; - std::atomic got_client_hello_{false}; - std::atomic got_goodbye_{false}; - std::atomic got_client_time_{false}; -}; - /// TCP relay that accepts one connection, sits on it without reading for delay_ms (the peer's /// WebSocket upgrade request waits in the kernel buffer), then connects to the backend and pumps /// bytes both ways. Simulates a slow network path in front of a real Sendspin server. @@ -380,7 +245,7 @@ TEST(ConnectionLifecycle, JunkProbeDoesNotBlockRealServer) { TestNetworkProvider network; SendspinClient client(make_config(PROBE_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server // Hold a raw TCP connection open without ever speaking WebSocket. @@ -440,7 +305,7 @@ TEST(ConnectionLifecycle, SlowOutboundSurvivesUpgradeTier) { TestNetworkProvider network; SendspinClient client(make_config(OUTBOUND_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server client.connect_to(server_url(PROXY_LISTEN_PORT)); @@ -479,7 +344,7 @@ TEST(ConnectionLifecycle, InFlightOutboundDoesNotBlockInboundAdmission) { TestNetworkProvider network; SendspinClient client(make_config(ADMIT_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server client.connect_to(server_url(STALL_LISTEN_PORT)); @@ -509,7 +374,7 @@ TEST(ConnectionLifecycle, EarlyServerHelloDoesNotWedge) { TestNetworkProvider network; SendspinClient client(make_config(EARLY_HELLO_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server FakeServer eager(server_url(EARLY_HELLO_TEST_PORT), "server-eager", {.hello_on_open = true}); @@ -530,7 +395,7 @@ TEST(ConnectionLifecycle, TwoServerRaceResolvedByPreference) { SendspinClient client(make_config(RACE_TEST_PORT)); client.set_network_provider(&network); client.set_persistence_provider(&persistence); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server // server-a establishes and is promoted into the empty slot first... @@ -561,7 +426,7 @@ TEST(ConnectionLifecycle, HeldProbesNeverOccupyNursery) { TestNetworkProvider network; SendspinClient client(make_config(EVICT_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server // Two held raw probes, enough to fill every nursery slot if they were admitted at accept. @@ -595,7 +460,7 @@ TEST(ConnectionLifecycle, FullNurseryOfLivePeersRejectsNewcomer) { TestNetworkProvider network; SendspinClient client(make_config(REJECT_TEST_PORT)); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server // Two mute peers: they upgrade and receive client/hello but never answer it, occupying both @@ -644,7 +509,7 @@ TEST(ConnectionLifecycle, SilentEstablishedPeerIsDropped) { config.liveness_timeout_ms = 300; SendspinClient client(config); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server FakeServer silent(server_url(LIVENESS_TEST_PORT), "server-silent", {.answer_time = false}); @@ -668,7 +533,7 @@ TEST(ConnectionLifecycle, AnsweringPeerSurvivesLivenessTimeout) { config.liveness_timeout_ms = 300; SendspinClient client(config); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server FakeServer live(server_url(LIVENESS_CONTROL_PORT), "server-live", {.answer_time = true}); @@ -694,7 +559,7 @@ TEST(ConnectionLifecycle, DisabledLivenessKeepsSilentPeer) { config.liveness_timeout_ms = 0; SendspinClient client(config); client.set_network_provider(&network); - ASSERT_TRUE(client.start_server()); + ASSERT_TRUE(client.start()); client.loop(); // First tick binds the WS server FakeServer silent(server_url(LIVENESS_DISABLED_PORT), "server-silent", {.answer_time = false}); diff --git a/tests/test_support.h b/tests/test_support.h new file mode 100644 index 00000000..90419a3f --- /dev/null +++ b/tests/test_support.h @@ -0,0 +1,216 @@ +// Copyright 2026 Sendspin Contributors +// +// 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. + +/// @file test_support.h +/// @brief Loopback scaffolding shared by the tests that drive a whole SendspinClient: an +/// IXWebSocket endpoint that plays the Sendspin server, a network provider that is always ready, +/// and pump helpers that tick client.loop() while waiting on a predicate. + +#pragma once + +#include "platform/time.h" +#include "sendspin/client.h" +#include "sendspin/config.h" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace sendspin::test { + +inline std::string server_url(uint16_t port) { + return "ws://127.0.0.1:" + std::to_string(port) + "/sendspin"; +} + +inline std::string server_hello_json(const std::string& server_id, const std::string& reason) { + return std::string(R"({"type":"server/hello","payload":{"server_id":")") + server_id + + R"(","name":"Fake Server","version":1,"active_roles":["player"],)" + + R"("connection_reason":")" + reason + R"("}})"; +} + +inline SendspinClientConfig make_config(uint16_t port) { + SendspinClientConfig config; + config.client_id = "lifecycle-test-client"; + config.name = "Lifecycle Test Client"; + config.server_port = port; + return config; +} + +class TestNetworkProvider : public SendspinNetworkProvider { +public: + bool is_network_ready() override { + return true; + } +}; + +/// Pumps client.loop() until pred() is true. No timeout: a regression hangs here and the suite +/// watchdog reports it. +inline void pump_until(SendspinClient& client, const std::function& pred) { + for (;;) { + client.loop(); + if (pred()) { + return; + } + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } +} + +/// Pumps client.loop() for a fixed window. Only for spacing events or "must not happen" checks: +/// a window that is too short can miss a regression, never fail a correct run. +inline void pump_for(SendspinClient& client, int duration_ms) { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(duration_ms); + while (std::chrono::steady_clock::now() < deadline) { + client.loop(); + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } +} + +/// Blocks until pred() is true without pumping the client (for checks on a stopped client). +inline void wait_until(const std::function& pred) { + while (!pred()) { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } +} + +/// Behavior knobs for FakeServer. +struct FakeServerOptions { + bool hello_on_open{false}; ///< Send server/hello immediately on Open, before any + ///< client/hello arrives (a nonconforming peer) + bool answer_hello{true}; ///< Reply to client/hello with server/hello (false: mute peer + ///< that upgrades and then never establishes) + bool answer_time{false}; ///< Reply to client/time with a server/time whose clock is the + ///< client's own (both sides read platform_time_us(), so the + ///< offset is ~0 and audio timestamps mean what they say) +}; + +/// A minimal Sendspin server: an IXWebSocket client that connects to the SendspinClient's WS +/// server (the server-initiated discovery direction), answers client/hello with server/hello per +/// the given options, and records the goodbye and close. +class FakeServer { +public: + FakeServer(const std::string& url, std::string server_id, FakeServerOptions options = {}) + : server_id_(std::move(server_id)) { + this->ws_.setUrl(url); + this->ws_.disableAutomaticReconnection(); + this->ws_.setOnMessageCallback([this, options](const ix::WebSocketMessagePtr& msg) { + if (msg->type == ix::WebSocketMessageType::Open) { + if (options.hello_on_open) { + this->ws_.send(server_hello_json(this->server_id_, "discovery")); + } + } else if (msg->type == ix::WebSocketMessageType::Message) { + const std::string& text = msg->str; + if (text.find("client/hello") != std::string::npos) { + this->got_client_hello_.store(true); + if (options.answer_hello) { + this->ws_.send(server_hello_json(this->server_id_, "discovery")); + } + } else if (text.find("client/time") != std::string::npos) { + this->got_client_time_.store(true); + if (options.answer_time) { + this->answer_time(text); + } + } else if (text.find("client/goodbye") != std::string::npos) { + { + std::lock_guard lock(this->goodbye_mutex_); + this->goodbye_message_ = text; + } + this->got_goodbye_.store(true); + } + } else if (msg->type == ix::WebSocketMessageType::Close || + msg->type == ix::WebSocketMessageType::Error) { + this->closed_.store(true); + } + }); + this->ws_.start(); + } + + ~FakeServer() { + this->ws_.stop(); + } + + void send_text(const std::string& text) { + this->ws_.send(text); + } + + /// Sends one binary message: type byte, big-endian server timestamp, then the payload. + void send_binary(uint8_t binary_type, int64_t timestamp_us, const std::string& payload) { + std::string frame; + frame.push_back(static_cast(binary_type)); + for (int shift = 56; shift >= 0; shift -= 8) { + frame.push_back(static_cast((timestamp_us >> shift) & 0xFF)); + } + frame.append(payload); + this->ws_.sendBinary(frame); + } + + /// Sends one player audio chunk: binary type 4 with a zeroed PCM payload. + void send_audio(int64_t timestamp_us, size_t payload_bytes) { + this->send_binary(4, timestamp_us, std::string(payload_bytes, '\0')); + } + + bool closed() const { + return this->closed_.load(); + } + + bool got_client_hello() const { + return this->got_client_hello_.load(); + } + + bool got_goodbye() const { + return this->got_goodbye_.load(); + } + + std::string goodbye_message() const { + std::lock_guard lock(this->goodbye_mutex_); + return this->goodbye_message_; + } + + bool got_client_time() const { + return this->got_client_time_.load(); + } + +private: + void answer_time(const std::string& client_time_text) { + const auto pos = client_time_text.find("\"client_transmitted\":"); + if (pos == std::string::npos) { + return; + } + const long long client_transmitted = + std::strtoll(client_time_text.c_str() + pos + 21, nullptr, 10); + const int64_t now = platform_time_us(); + this->ws_.send(std::string(R"({"type":"server/time","payload":{)") + + "\"client_transmitted\":" + std::to_string(client_transmitted) + + ",\"server_received\":" + std::to_string(now) + + ",\"server_transmitted\":" + std::to_string(now) + "}}"); + } + + ix::WebSocket ws_; + std::string server_id_; + mutable std::mutex goodbye_mutex_; + std::string goodbye_message_; + std::atomic closed_{false}; + std::atomic got_client_hello_{false}; + std::atomic got_goodbye_{false}; + std::atomic got_client_time_{false}; +}; + +} // namespace sendspin::test