From 09cc38b01ca9ffe46a31829359102aaea88eaf13 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Mon, 31 Aug 2026 20:52:16 -0500 Subject: [PATCH 1/2] Add a source capture host example and sync the documentation --- CLAUDE.md | 9 +- CMakeLists.txt | 7 + README.md | 5 +- docs/conventions.md | 5 +- docs/integration-guide.md | 97 +++++- docs/internals.md | 70 +++- examples/source_client/CMakeLists.txt | 40 +++ examples/source_client/README.md | 48 +++ examples/source_client/main.cpp | 484 ++++++++++++++++++++++++++ 9 files changed, 747 insertions(+), 18 deletions(-) create mode 100644 examples/source_client/CMakeLists.txt create mode 100644 examples/source_client/README.md create mode 100644 examples/source_client/main.cpp diff --git a/CLAUDE.md b/CLAUDE.md index 9e1a074..be7d7da 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,9 @@ The library provides `SendspinClient` as the main public API. It handles the ful - `ArtworkRole` (`artwork_role.h`): receives album artwork images - `VisualizerRole` (`visualizer_role.h`): receives spectrum/beat visualization data - `ColorRole` (`color_role.h`): receives audio-derived RGB color palette from the server +- `SourceRole` (`source_role.h`): audio capture role, owns `SourceTask`, accepts captured audio via `write_audio()` and streams it to the server - `SyncTask` (`sync_task.h`): decodes encoded audio, synchronizes to server timestamps, writes PCM via audio write callback +- `SourceTask` (`source_task.h`): assembles captured audio into timestamped chunks, encodes them via `SourceEncoder` (PCM passthrough or Opus), and sends them on the stream's bound connection - `SendspinConnection` (`connection.h`): abstract WebSocket connection base - `SendspinServerConnection` / `SendspinClientConnection`: platform-specific WebSocket transports (ESP uses `esp_websocket_client`/`esp_http_server`, host uses IXWebSocket) - `Inbox` / `InboxSlot` (`inbox.h`): single-mutex mailbox for all main-loop-bound cross-thread state - atomic topic bitmask polled lock-free by `loop()`, plus a fixed event ring for ordered lifecycle/time events @@ -25,9 +27,9 @@ The library provides `SendspinClient` as the main public API. It handles the ful ### Role composition -Roles are added to the client at runtime via `add_player()`, `add_metadata()`, etc. Each role receives a `SendspinClient*` at construction time and uses it to access shared services (time sync, state publishing, message sending). The consumer provides behavior by implementing listener interfaces (`PlayerRoleListener`, `MetadataRoleListener`, etc.) and setting them via `set_listener()`. Required callbacks are pure virtual; optional callbacks have default no-op implementations. The client dispatches messages to roles via null-pointer checks on role pointers. +Roles are added to the client at runtime via `add_player()`, `add_source()`, `add_metadata()`, etc. Each role receives a `SendspinClient*` at construction time and uses it to access shared services (time sync, state publishing, message sending). The consumer provides behavior by implementing listener interfaces (`PlayerRoleListener`, `MetadataRoleListener`, etc.) and setting them via `set_listener()`. Required callbacks are pure virtual; optional callbacks have default no-op implementations. The client dispatches messages to roles via null-pointer checks on role pointers. -Roles can be disabled at compile time via `SENDSPIN_ENABLE_*` cmake options (host build) or Kconfig entries (ESP-IDF build). When a role is disabled, its source files are not compiled and its `add_*()` declaration, accessor, and `unique_ptr` member are removed from `client.h`. `#ifdef` guards live in exactly two places in the library: `cmake/sources.cmake` (source lists) and `include/sendspin/client.h` / `src/client.cpp` (dispatch points); examples guard their own role usage like any consumer. Audio codec dependencies (micro-flac, micro-opus) are only linked when the player role is enabled. +Roles can be disabled at compile time via `SENDSPIN_ENABLE_*` cmake options (host build) or Kconfig entries (ESP-IDF build). When a role is disabled, its source files are not compiled and its `add_*()` declaration, accessor, and `unique_ptr` member are removed from `client.h`. `#ifdef` guards live in exactly two places in the library: `cmake/sources.cmake` (source lists) and `include/sendspin/client.h` / `src/client.cpp` (dispatch points); examples guard their own role usage like any consumer (a single-role example may instead gate its whole target in CMake). Audio codec dependencies follow the roles that use them: micro-flac is linked only when the player role is enabled; micro-opus is linked when the player or source role is enabled. The consuming platform (e.g., ESPHome) supplies the listener implementations plus `SendspinNetworkProvider` and the optional `SendspinPersistenceProvider`/`SendspinClientListener` providers; `docs/integration-guide.md` has the full wiring, including a minimal working example. @@ -43,6 +45,7 @@ cmake/ - CMake modules (sources.cmake, host.cmake) examples/common/ - Shared PortAudio audio sink used by host examples examples/basic_client/ - Standalone host example with PortAudio audio output examples/tui_client/ - Terminal UI host example with PortAudio audio output +examples/source_client/ - Host example streaming PortAudio input capture as a source tests/ - Host unit tests (GoogleTest) docs/ - integration-guide.md (consumer guide), internals.md (how the current code works), conventions.md (normative design standards) .claude/skills/ - Review checklists applying the standards to a diff (docs-sync, embedded-review, house-patterns, test-standards) @@ -50,7 +53,7 @@ docs/ - integration-guide.md (consumer guide), internals.m ### Header visibility -- **Public** (`include/sendspin/`): `client.h`, `config.h`, `types.h`, and role headers (`player_role.h`, `controller_role.h`, `metadata_role.h`, `artwork_role.h`, `visualizer_role.h`, `color_role.h`). These are the consumer-facing API. `config.h` contains all configuration structs (`SendspinClientConfig` and role configs). Each role header defines its own protocol types (enums, structs, conversion functions). `types.h` contains shared types used across the client and roles. +- **Public** (`include/sendspin/`): `client.h`, `config.h`, `types.h`, and role headers (`player_role.h`, `controller_role.h`, `metadata_role.h`, `artwork_role.h`, `visualizer_role.h`, `color_role.h`, `source_role.h`). These are the consumer-facing API. `config.h` contains all configuration structs (`SendspinClientConfig` and role configs). Each role header defines its own protocol types (enums, structs, conversion functions). `types.h` contains shared types used across the client and roles. - **Private** (`src/`): All internal headers (decoder, sync_task, time_filter, ring buffers, protocol_messages, etc.). Not exposed to consumers. `protocol_messages.h` contains message envelope structs, internal protocol enums, and protocol function declarations. - **Platform-specific** (`src/esp/`, `src/host/`): Networking headers with the same names (`client_connection.h`, `server_connection.h`, `ws_server.h`) but different implementations per platform. diff --git a/CMakeLists.txt b/CMakeLists.txt index 7d9a008..a04d291 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -206,6 +206,13 @@ else() if(BUILD_EXAMPLES) add_subdirectory(examples/basic_client) add_subdirectory(examples/tui_client) + # source_client exists solely to demonstrate the source role, so its whole target is + # gated on the role's option (docs/conventions.md, Public API) + if(SENDSPIN_ENABLE_SOURCE) + add_subdirectory(examples/source_client) + else() + message(STATUS "Skipping source_client example (SENDSPIN_ENABLE_SOURCE=OFF)") + endif() else() message(STATUS "Skipping sendspin-cpp examples (BUILD_EXAMPLES=OFF)") endif() diff --git a/README.md b/README.md index 291955c..cce10b1 100644 --- a/README.md +++ b/README.md @@ -9,9 +9,9 @@ Standalone C++ library implementing the [Sendspin synchronized audio streaming p ## Features -- Modular Sendspin role composition: artwork, color, controller, metadata, player, and visualizer +- Modular Sendspin role composition: artwork, color, controller, metadata, player, source, and visualizer - WebSocket client and server support -- Decodes FLAC, Opus, and PCM +- Decodes FLAC, Opus, and PCM for playback; encodes Opus and PCM for source capture - Cross-platform: ESP-IDF (ESP32) and host (macOS/Linux) ## Documentation @@ -45,6 +45,7 @@ Requires ESP-IDF v5.1 or later. - **`examples/basic_client/`** -- Standalone host example with PortAudio audio output - **`examples/tui_client/`** -- Terminal UI host example with PortAudio audio output +- **`examples/source_client/`** -- Standalone host example streaming PortAudio input capture as a source ## License diff --git a/docs/conventions.md b/docs/conventions.md index d6c56c8..b901826 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -96,7 +96,10 @@ checklists in `.claude/skills/` apply these standards to a diff. at the declaration site. - Examples are consumers too: they must build under every `SENDSPIN_ENABLE_*` combination, guarding role usage the same way an - external consumer would. + external consumer would. An example that exists solely to demonstrate one + role may instead gate its whole target in CMake on that role's option; + an `#ifdef` branch that leaves a do-nothing binary is dead code, not a + guard. ## Consistency diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 7fcc615..a7a13a6 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -7,7 +7,7 @@ This guide describes what you need to implement in order to integrate sendspin-c Integration follows this pattern: 1. Create a `SendspinClient` with a configuration struct -2. Add roles (player, controller, metadata, artwork, visualizer, color) depending on what your application needs +2. Add roles (player, controller, metadata, artwork, visualizer, color, source) depending on what your application needs 3. Implement listener interfaces for the roles you added 4. Implement a network provider (required) and optionally a persistence provider 5. Wire listeners and providers to the client and roles @@ -27,6 +27,7 @@ Include `sendspin/client.h` for the client class, config types, and shared types #include "sendspin/artwork_role.h" // ArtworkRole, ArtworkRoleListener #include "sendspin/visualizer_role.h" // VisualizerRole, VisualizerRoleListener #include "sendspin/color_role.h" // ColorRole, ColorRoleListener +#include "sendspin/source_role.h" // SourceRole, SourceRoleListener ``` Only include the role headers you need. `client.h` includes `sendspin/config.h` (all configuration structs, including `SendspinClientConfig`) and `sendspin/types.h` transitively. @@ -169,6 +170,43 @@ Receives an RGB color palette derived by the server from the currently playing a auto& color = client.add_color(); ``` +### Source Role (Audio Capture) + +Streams audio captured by the client (e.g., a line input or microphone) to the server. Requires a configuration struct that declares the capture format; that format is fixed for the lifetime of the role — the server does not negotiate it, and changing formats requires tearing down the client and re-adding the role with a new config. + +```cpp +SourceRoleConfig source_config; +source_config.codec = SendspinCodecFormat::PCM; // or OPUS to encode before sending +source_config.sample_rate = 48000; +source_config.channels = 2; +source_config.bit_depth = 16; + +auto& source = client.add_source(source_config); +``` + +The configuration is validated at `add_source()` time; validation fails closed. An invalid config (see the [SourceRoleConfig](#sourceroleconfig) reference for the per-field rules) logs each rejected field at ERROR and leaves the role inert: it is not advertised to the server and never streams, exactly like a player with no audio formats. + +Streaming is gated by the server: the client never streams unsolicited, the default after connect is stopped, and permission does not survive reconnection. When the server commands start, the role opens the outbound stream and fires `on_streaming_started()`; from that point on, feed captured audio to `write_audio()`: + +```cpp +// Capture thread (exactly one producer thread): +source.write_audio(pcm_bytes, len, capture_time_us); +``` + +- `data`/`len`: interleaved little-endian signed PCM in the configured format (24-bit as 3 packed bytes per sample). `len` must be a whole number of frames (one frame = one sample across all channels); a non-whole-frame write is rejected as a whole. +- `capture_time_us`: local-clock capture time of the FIRST sample in the buffer, in the same domain as the client's time functions (`std::chrono::steady_clock` microseconds on host). Pass `0` to stamp with the current time, a best-effort fallback for callers that cannot timestamp their ADC. +- Returns `true` if the audio was accepted; `false` when the stream is not open or the capture buffer is full (the write is dropped and counted). + +`write_audio()` is a non-blocking, non-allocating hot path, safe to call from an audio capture callback. Exactly one producer thread may call it; the library does not serialize concurrent writers. + +If the capture hardware supports line-input signal sensing, set `SourceRoleConfig::line_sense` and report the signal state from the main loop thread; the role publishes it to the server via `client/state`: + +```cpp +source.set_signal(SourceSignal::PRESENT); // or ABSENT +``` + +`is_streaming()` (main loop thread only) returns `true` between the `on_streaming_started()` and `on_streaming_stopped()` callbacks. + ## Step 3: Implement Listener Interfaces ### PlayerRoleListener (Required if Using Player Role) @@ -416,6 +454,21 @@ The `ServerColorStateObject` contains a `timestamp` and six optional `RgbColor` A field is `nullopt` when the server has not provided it or has explicitly cleared it; listeners do not need to distinguish those cases. +### SourceRoleListener + +Both callbacks are optional and fire on the main loop thread. They bracket the outbound stream: enable your capture path in `on_streaming_started()` and disable it in `on_streaming_stopped()` (`write_audio()` rejects audio outside that window either way). + +```cpp +struct MySourceListener : SourceRoleListener { + // Called when the outbound stream to the server has opened. + void on_streaming_started() override { capture.enable(); } + + // Called when the outbound stream has closed (server stop command, + // connection loss, or disconnect). + void on_streaming_stopped() override { capture.disable(); } +}; +``` + ## Step 4: Implement Providers ### SendspinNetworkProvider (Required) @@ -590,6 +643,7 @@ if (auto* m = client.metadata()) { if (auto* a = client.artwork()) { /* ... */ } if (auto* v = client.visualizer()) { /* ... */ } if (auto* col = client.color()) { /* ... */ } +if (auto* s = client.source()) { /* ... */ } ``` Use these accessors when the role reference from `add_*()` is out of scope. @@ -664,6 +718,8 @@ Most listener callbacks fire on the main loop thread (the thread calling `client `ArtworkRole::frame_done()` must be called from the main loop thread (typically from inside `on_image_display()`/`on_image_clear()` or when a cross-fade animation completes). +`SourceRole::write_audio()` is designed to be called from your audio capture thread — exactly one producer thread; the library does not serialize concurrent writers. The other `SourceRole` methods (`set_signal()`, `is_streaming()`) must be called from the main loop thread. + ## Minimal Example A minimal integration that receives and discards audio: @@ -726,7 +782,8 @@ cmake -B build -DSENDSPIN_ENABLE_CONTROLLER=OFF \ -DSENDSPIN_ENABLE_METADATA=OFF \ -DSENDSPIN_ENABLE_ARTWORK=OFF \ -DSENDSPIN_ENABLE_VISUALIZER=OFF \ - -DSENDSPIN_ENABLE_COLOR=OFF + -DSENDSPIN_ENABLE_COLOR=OFF \ + -DSENDSPIN_ENABLE_SOURCE=OFF ``` Available options (all `ON` by default): @@ -739,8 +796,9 @@ Available options (all `ON` by default): | `SENDSPIN_ENABLE_ARTWORK` | Artwork role | | `SENDSPIN_ENABLE_VISUALIZER` | Visualizer role | | `SENDSPIN_ENABLE_COLOR` | Color role | +| `SENDSPIN_ENABLE_SOURCE` | Source role, Opus encoder (micro-opus), source task | -When `SENDSPIN_ENABLE_PLAYER` is `OFF`, the micro-flac and micro-opus dependencies are not fetched. +The codec dependencies follow the roles that use them: micro-flac is fetched only when `SENDSPIN_ENABLE_PLAYER` is `ON`; micro-opus is fetched when `SENDSPIN_ENABLE_PLAYER` or `SENDSPIN_ENABLE_SOURCE` is `ON`. ### ESP-IDF (Kconfig) @@ -753,6 +811,7 @@ CONFIG_SENDSPIN_ENABLE_METADATA=y CONFIG_SENDSPIN_ENABLE_ARTWORK=y CONFIG_SENDSPIN_ENABLE_VISUALIZER=y CONFIG_SENDSPIN_ENABLE_COLOR=y +CONFIG_SENDSPIN_ENABLE_SOURCE=y ``` ### Effect on the API @@ -868,6 +927,27 @@ Configuration passed to `client.add_visualizer()`. --- +### SourceRoleConfig + +Configuration passed to `client.add_source()`. It is the capture format contract for every stream the role opens. Validation fails closed at `add_source()` time: any rule violation below logs at ERROR and leaves the role inert (not advertised, never streaming) — values are never clamped or repaired. + +| Field | Type | Default | Description | +|---|---|---|---| +| `codec` | `SendspinCodecFormat` | `PCM` | Outbound codec: `PCM` (chunks are the capture bytes, untouched) or `OPUS` (each chunk is encoded into one Opus packet). Any other value is rejected. `OPUS` costs the encoder state plus micro-opus's per-thread scratch arena (~120 KB, SPIRAM-preferred, allocated lazily on the source task's first encode) — the same per-thread arena the player's Opus decode allocates on its own task, so a device doing both holds two such arenas. | +| `sample_rate` | `uint32_t` | `48000` | Capture sample rate in Hz; must be > 0. `OPUS` accepts only libopus's rates: 8000, 12000, 16000, 24000, or 48000 (a 44100 line-in must use `PCM` or resample upstream). | +| `channels` | `uint8_t` | `2` | Capture channel count; must be > 0 (1 or 2 for `OPUS`). | +| `bit_depth` | `uint8_t` | `16` | Bits per sample: 16, 24 (3 packed bytes per sample), or 32. `OPUS` requires 16. | +| `chunk_duration_ms` | `uint32_t` | `25` | Duration of one outbound audio chunk. `PCM` accepts the spec bounds [5, 150]; `OPUS` accepts only 10, 20, 40, or 60 (one chunk is exactly one legal Opus frame), so the PCM default of 25 is rejected for `OPUS`. | +| `capture_buffer_ms` | `uint32_t` | `150` | Capture ring capacity in milliseconds of audio in the configured format; must be > 0. The ring is the stall-policy backlog bound: capture beyond it is dropped at `write_audio()` and streaming resumes from live audio. Approximate: per-write ring metadata comes out of a fixed +25% margin, so many very small `write_audio()` calls reduce the effective audio capacity below this figure. | +| `opus_bitrate` | `uint32_t` | `128000` | Opus bitrate in bit/s, validated against [500, 512000] (the range libopus accepts). Ignored (and unvalidated) when `codec` is `PCM`. | +| `opus_complexity` | `uint8_t` | `2` | Opus encoder complexity, validated to at most 10. The low default fits an ESP32-class real-time encode budget; hosts may raise it for quality per CPU. Ignored (and unvalidated) when `codec` is `PCM`. | +| `line_sense` | `bool` | `false` | Advertise line-input signal sensing to the server; see `SourceRole::set_signal()`. | +| `priority` | `unsigned` | `3` | FreeRTOS priority for the source task (ESP-IDF only). Below the HTTP server task (`5`) and the sync/decode task (`6`) so outbound capture can never starve inbound playback. | +| `buffer_location` | `MemoryLocation` | `PREFER_EXTERNAL` | Memory placement for the capture ring, chunk staging buffer, and the Opus encoder's scratch buffers. ESP-IDF only; ignored on host. | +| `psram_stack` | `bool` | `false` | Allocate source task stack in PSRAM (ESP-IDF only) | + +--- + ## Enums Reference ### SendspinCodecFormat @@ -941,6 +1021,15 @@ These represent commands the server can send to the player. The player advertise | `ONE` | Repeat current track | | `ALL` | Repeat all tracks | +### SourceSignal + +| Value | Description | +|---|---| +| `PRESENT` | Audio signal detected on the capture input | +| `ABSENT` | No audio signal on the capture input | + +Reported via `SourceRole::set_signal()`; only meaningful when `SourceRoleConfig::line_sense` is set. + ### SendspinImageFormat | Value | Description | @@ -995,4 +1084,4 @@ Set with `SendspinClient::set_log_level()`. Only affects host builds; ESP-IDF bu | `PREFER_EXTERNAL` | Prefer SPIRAM, fall back to internal RAM (ESP-IDF only) | | `PREFER_INTERNAL` | Prefer internal RAM, fall back to SPIRAM (ESP-IDF only) | -Used by `SendspinClientConfig::websocket_payload_location` to control where the per-connection WebSocket payload reassembly buffer is allocated, and by `PlayerRoleConfig::decode_buffer_location` to control where the player's decode transfer buffer is allocated. Ignored on host platforms (no internal/external distinction). +Used by `SendspinClientConfig::websocket_payload_location` to control where the per-connection WebSocket payload reassembly buffer is allocated, by `PlayerRoleConfig::decode_buffer_location` to control where the player's decode transfer buffer is allocated, and by `SourceRoleConfig::buffer_location` to control where the source's capture ring, chunk staging buffer, and Opus encoder scratch buffers are allocated. Ignored on host platforms (no internal/external distinction). diff --git a/docs/internals.md b/docs/internals.md index 2dc016c..167a77e 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -14,7 +14,7 @@ Throughout this document, internal field and method references use the `Impl` qu Roles can be disabled at build time via `SENDSPIN_ENABLE_*` (CMake options on host, Kconfig entries on ESP-IDF). Two mechanisms cooperate, with a strict boundary between them: -1. **CMake source-list exclusion** (`cmake/sources.cmake`). Each role has its own `SENDSPIN__SOURCES` list. When a role is disabled, its translation units are not added to the build, so the code never compiles and its transitive dependencies are not required; e.g., micro-flac and micro-opus for the player. The ESP-IDF component manifest (`idf_component.yml`) similarly gates the audio codec dependencies on `SENDSPIN_ENABLE_PLAYER` so they are not even fetched. +1. **CMake source-list exclusion** (`cmake/sources.cmake`). Each role has its own `SENDSPIN__SOURCES` list. When a role is disabled, its translation units are not added to the build, so the code never compiles and its transitive dependencies are not required; e.g., micro-flac for the player and micro-opus for the player and source. The ESP-IDF component manifest (`idf_component.yml`) similarly gates the codec dependencies so they are not even fetched: micro-flac on `SENDSPIN_ENABLE_PLAYER`, micro-opus on `SENDSPIN_ENABLE_PLAYER` or `SENDSPIN_ENABLE_SOURCE`. 2. **`#ifdef SENDSPIN_ENABLE_` guards** in `include/sendspin/client.h` and `src/client.cpp`. These are the only core files that must reference role types directly (the `std::unique_ptr` members, `add_*()` / accessor declarations, and dispatch branches in message handlers). Nowhere else in the core should use these guards. The split exists because the two problems are different. CMake handles "don't compile this file and don't require its dependencies," while `#ifdef` handles "core code needs to conditionally mention a type." Using `#ifdef` to gate entire files would still force the codec headers onto the include path; using CMake to gate individual member declarations is not possible. @@ -34,9 +34,10 @@ The library uses a small number of long-lived threads. All state mutations and u | Thread | Name | Created by | Stack (ESP) | Priority (ESP) | Purpose | |--------|------|-----------|-------------|-----------------|---------| | **Main loop** | (caller's) | User code | - | - | Drives `SendspinClient::loop()`. All role event processing and listener callbacks run here. | -| **Sync task** | `Sendspin` | `PlayerRole::Impl::start()` → `SyncTask::start()` | 6192 B | 2 | Decodes audio, synchronizes to server timestamps, writes PCM to the audio sink via `on_audio_write`. | +| **Sync task** | `Sendspin` | `PlayerRole::Impl::start()` → `SyncTask::start()` | 6192 B | 6 | Decodes audio, synchronizes to server timestamps, writes PCM to the audio sink via `on_audio_write`. | | **Visualizer drain** | `SsVis` | `VisualizerRole::Impl::start()` | 4096 B | 2 | Reads visualization frames from a ring buffer and delivers them to the listener at the correct playback time. | | **Artwork decode** | `SsArt` | `ArtworkRole::Impl::start()` | 4096 B | 2 | Receives image notifications and calls the decode callback. Hands the server display timestamp off to the main loop, which fires the display callback at the correct time. | +| **Source task** | `SsSrc` | `SourceRole::Impl::start()` → `SourceTask::start()` | 6192 B | 3 | Drains the capture ring, assembles timestamped chunks, optionally Opus-encodes them, and sends them (plus the stream's `client-stream/start`/`end` messages) on the bound connection. | | **Network** | (library-internal) | IXWebSocket (host) or esp_http_server (ESP) | - | - | WebSocket I/O. Callbacks fire on these threads and must defer work to the main loop. | On host builds, `platform_configure_thread()` is a no-op; threads use OS defaults. On ESP-IDF it calls `esp_pthread_set_cfg()` to set stack size, priority, name, and optional PSRAM allocation before the `std::thread` is constructed. @@ -56,6 +57,12 @@ On host builds, `platform_configure_thread()` is a no-op; threads use OS default 2. The thread blocks on ring buffer receives with a 50 ms timeout. 3. `VisualizerRole::Impl` destructor sets `COMMAND_STOP` and joins. +**Source task** (`src/source_task.cpp`): + +1. `SourceRole::Impl::start()` calls `SourceTask::init()` (allocates the capture ring and chunk staging buffer, creates the encoder) and `SourceTask::start()`, which spawns the thread and blocks until it reaches IDLE (`SOURCE_TASK_IDLE`). +2. The thread runs a persistent outer loop for the lifetime of the client, idling between streams (see [Source Streaming Pipeline](#source-streaming-pipeline)). +3. `SourceTask::stop()` sets `SOURCE_COMMAND_STOP` and joins. Called from `SourceTask`'s destructor, triggered by `SourceRole::Impl`'s destructor. + **Artwork decode** (`src/artwork_role.cpp`): 1. `ArtworkRole::Impl::start()` spawns the decode thread. @@ -84,7 +91,7 @@ TASK_ERROR (1 << 11) Allocation or decode failure TASK_IDLE (1 << 12) Waiting for work ``` -The sync task, visualizer drain thread, and artwork decode thread all use event flags for command signaling from the main loop and status reporting back. The artwork decode thread uses a simpler subset: just `COMMAND_STOP`. The visualizer drain thread adds `COMMAND_FLUSH` and `COMMAND_CLEAR`. `COMMAND_CLEAR` discards buffered entries up to a 1-byte marker the network thread enqueues on `stream/start` and `stream/clear` (mirroring the sync task's clear-marker chunk) so frames received after the boundary survive; `COMMAND_FLUSH` (drain to empty) is only used when the producer is already stopped (`stream/end`, cleanup). +The sync task, visualizer drain thread, artwork decode thread, and source task all use event flags for command signaling from the main loop and status reporting back. The artwork decode thread uses a simpler subset: just `COMMAND_STOP`. The visualizer drain thread adds `COMMAND_FLUSH` and `COMMAND_CLEAR`. `COMMAND_CLEAR` discards buffered entries up to a 1-byte marker the network thread enqueues on `stream/start` and `stream/clear` (mirroring the sync task's clear-marker chunk) so frames received after the boundary survive; `COMMAND_FLUSH` (drain to empty) is only used when the producer is already stopped (`stream/end`, cleanup). The source task has its own enum (`SourceTaskBits` in `src/source_task.h`): `SOURCE_COMMAND_STOP`/`SOURCE_TASK_RUNNING`/`SOURCE_TASK_STOPPED`/`SOURCE_TASK_IDLE` mirror the sync task's lifecycle bits, `SOURCE_COMMAND_UPDATE` signals that the desired streaming state (a latest-wins atomic) changed, and `SOURCE_SEND_COMPLETE` paces chunk sends against their completion callbacks. ### ThreadSafeQueue (`src/platform/thread_safe_queue.h`) @@ -117,7 +124,7 @@ State currently on the Inbox: | Endpoint | Topic bit | Data | Producer | |----------|-----------|------|----------| -| Event ring | `INBOX_TOPIC_EVENTS` | Lifecycle events (`TimeResponsePayload`, `PLAYER_STREAM` STREAM_START/STREAM_END, `ARTWORK_STREAM` STREAM_END/STREAM_CLEAR, `VISUALIZER_STREAM` STREAM_START/STREAM_END/STREAM_CLEAR, plus `CONTROLLER_CLEARED` / `METADATA_CLEARED` / `COLOR_CLEARED`) via `InboxEvent` | Network thread (`TimeResponsePayload`, `PLAYER_STREAM`, `ARTWORK_STREAM`, `VISUALIZER_STREAM`) / main-loop thread (`*_CLEARED` and the synthetic `cleanup()` stream events) | +| Event ring | `INBOX_TOPIC_EVENTS` | Lifecycle events (`TimeResponsePayload`, `PLAYER_STREAM` STREAM_START/STREAM_END, `ARTWORK_STREAM` STREAM_END/STREAM_CLEAR, `VISUALIZER_STREAM` STREAM_START/STREAM_END/STREAM_CLEAR, `SOURCE_STREAM` STREAMING_STARTED/STREAMING_STOPPED, plus `CONTROLLER_CLEARED` / `METADATA_CLEARED` / `COLOR_CLEARED`) via `InboxEvent` | Network thread (`TimeResponsePayload`, `PLAYER_STREAM`, `ARTWORK_STREAM`, `VISUALIZER_STREAM`) / source task thread (`SOURCE_STREAM`) / main-loop thread (`*_CLEARED` and the synthetic `cleanup()` stream events) | | `Client::group_slot` | `INBOX_TOPIC_GROUP` | `GroupUpdateObject` (field-by-field delta merge) | Network thread | | `ControllerRole::Impl::slot` | `INBOX_TOPIC_CONTROLLER` | `ServerStateControllerObject` (latest wins) | Network thread | | `MetadataRole::Impl::slot` | `INBOX_TOPIC_METADATA` | `ServerMetadataStateDelta` (field-by-field delta merge) | Network thread | @@ -127,8 +134,9 @@ State currently on the Inbox: | `PlayerRole::Impl::EventState::state_slot` | `INBOX_TOPIC_PLAYER_STATE` | `SendspinClientState` (latest wins) | Sync task thread | | `VisualizerRole::Impl::EventState::config_slot` | `INBOX_TOPIC_VISUALIZER_CONFIG` | `ServerVisualizerStreamObject` (latest wins) | Network thread | | `ArtworkRole::Impl::EventState::display_slot` | `INBOX_TOPIC_ARTWORK_DISPLAY` | `ArtworkDisplayUpdate` (per-slot display timestamp + epoch, merged) | Artwork decode thread | +| `SourceRole::Impl::EventState::command_slot` | `INBOX_TOPIC_SOURCE_COMMAND` | `SourceCommandEnvelope` (server start/stop command + originating connection instance id, latest wins) | Network thread | -All roles have been migrated onto the Inbox. The controller/metadata/color roles write or merge server state into their `InboxSlot` from `handle_server_state()`, and their disconnect clear arrives as a `*_CLEARED` lifecycle event on the shared ring rather than a per-role flag. The player role owns three `InboxSlot`s (stream params, command, and client state, on its `EventState`) plus `PLAYER_STREAM` lifecycle events on the shared ring; its disconnect clear is the synthetic STREAM_END that `cleanup()` pushes onto the ring. The visualizer role writes its stream config to `config_slot` and delivers STREAM_START/END/CLEAR as `VISUALIZER_STREAM` ring events (no per-tick `drain_events()`; the config is taken when the START event is dispatched). The artwork role merges per-slot display timestamps (tagged with the decode `stream_epoch`) into `display_slot`, delivers STREAM_END/CLEAR as `ARTWORK_STREAM` ring events, and keeps a per-tick `drain_events()` for its server-clock display-deadline sweep. Both roles' disconnect clears are synthetic stream events that `cleanup()` pushes onto the ring. +All roles have been migrated onto the Inbox. The controller/metadata/color roles write or merge server state into their `InboxSlot` from `handle_server_state()`, and their disconnect clear arrives as a `*_CLEARED` lifecycle event on the shared ring rather than a per-role flag. The player role owns three `InboxSlot`s (stream params, command, and client state, on its `EventState`) plus `PLAYER_STREAM` lifecycle events on the shared ring; its disconnect clear is the synthetic STREAM_END that `cleanup()` pushes onto the ring. The visualizer role writes its stream config to `config_slot` and delivers STREAM_START/END/CLEAR as `VISUALIZER_STREAM` ring events (no per-tick `drain_events()`; the config is taken when the START event is dispatched). The artwork role merges per-slot display timestamps (tagged with the decode `stream_epoch`) into `display_slot`, delivers STREAM_END/CLEAR as `ARTWORK_STREAM` ring events, and keeps a per-tick `drain_events()` for its server-clock display-deadline sweep. Both roles' disconnect clears are synthetic stream events that `cleanup()` pushes onto the ring. The source role's `command_slot` is latest-wins on purpose: start and stop are idempotent and only the final desired state matters, so commands coalescing between drains loses nothing; its STREAMING_STARTED/STOPPED lifecycle callbacks ride the shared ring as `SOURCE_STREAM` events pushed by the source task thread. ### SpscRingBuffer (`src/platform/spsc_ring_buffer.h`) @@ -138,6 +146,7 @@ Used for: - **Encoded audio**: Via the `SendspinAudioRingBuffer` wrapper (which adds chunk headers and exposes `write_chunk` / `receive_chunk` / `return_chunk`). Network thread writes chunks; sync task reads and decodes them. - **Visualizer frames**: Used directly. Network thread writes one entry per visualizer binary message; drain thread reads them at the correct playback time. +- **Captured audio**: Via the same `SendspinAudioRingBuffer` wrapper, in the outbound direction. The consumer's capture thread writes one timestamped entry per `SourceRole::write_audio()` call; the source task reads entries and assembles them into wire chunks. ### Other Primitives @@ -176,7 +185,7 @@ Network thread (IXWebSocket / esp_http_server) | `SERVER_HELLO` | Stores server info and connection reason on the connection, then sets `server_hello_received_` (an atomic store that publishes the fields to the main loop; the manager's promotion scan observes `is_handshake_complete()` on its next tick) | | `SERVER_TIME` | Pushes a `TIME_RESPONSE` `InboxEvent` onto the shared inbox ring | | `SERVER_STATE` | Writes/merges into the controller, metadata, and color `InboxSlot`s via each role's `handle_server_state()` | -| `SERVER_COMMAND` | Merges into the player's `command_slot` (`InboxSlot`) | +| `SERVER_COMMAND` | Merges into the player's `command_slot` (`InboxSlot`); a source start/stop command is written to the source's `command_slot` (`InboxSlot`, latest wins) together with the originating connection's instance id | | `GROUP_UPDATE` | Merges into `Client::group_slot` (`InboxSlot`) | | `STREAM_START` | Writes to the player's `stream_params_slot`, pushes a `PLAYER_STREAM` (STREAM_START) event onto the inbox ring. Marks the artwork stream active, flushes the decode thread's notification queue, bumps the artwork `stream_epoch`, and resets the artwork `display_slot`. Writes the config to the visualizer's `config_slot` and pushes a `VISUALIZER_STREAM` (STREAM_START) event onto the inbox ring. | | `STREAM_END` | Pushes a `PLAYER_STREAM` (STREAM_END) event onto the inbox ring and signals sync task `COMMAND_STREAM_END`; pushes `ARTWORK_STREAM` (STREAM_END) and `VISUALIZER_STREAM` (STREAM_END) events onto the inbox ring | @@ -198,6 +207,8 @@ The bump arena suits ArduinoJson's allocation pattern: during a parse the varian | Artwork image | `ArtworkRole::Impl::handle_binary()`: copies image data to a per-slot double buffer and enqueues a notification for the artwork decode thread | | Visualizer data (binary types 16-20) | `VisualizerRole::Impl::handle_binary()`: writes to visualizer ring buffer | +The source role's binary ID block (types 12-15, `SENDSPIN_ROLE_SOURCE`) is outbound only: the client sends `SENDSPIN_BINARY_SOURCE_AUDIO` (12) chunks to the server and never receives messages in that block, so it has no inbound dispatch row (see [Outbound send path](#outbound-send-path)). + ### Main Loop Processing `SendspinClient::loop()` (`src/client.cpp`) runs the following steps **in order** on each tick: @@ -224,14 +235,17 @@ The bump arena suits ArduinoJson's allocation pattern: during a parse the varian ├─ Fire CONTROLLER/METADATA/COLOR_CLEARED via each role's handle_cleared_event() ├─ Dispatch PLAYER_STREAM via player_->impl_->on_stream_ring_event() ├─ Dispatch ARTWORK_STREAM via artwork_->impl_->handle_stream_ring_event() - └─ Dispatch VISUALIZER_STREAM via visualizer_->impl_->handle_stream_ring_event() + ├─ Dispatch VISUALIZER_STREAM via visualizer_->impl_->handle_stream_ring_event() + └─ Dispatch SOURCE_STREAM via source_->impl_->on_stream_ring_event() + (appends to pending_events; the callbacks fire from drain_events() below) 4. Role event draining (each role's impl_->drain_events(), gated on impl_->needs_drain(slot_bits)) ├─ player_->impl_->drain_events() ├─ controller_->impl_->drain_events() ├─ metadata_->impl_->drain_events() ├─ color_->impl_->drain_events() - └─ artwork_->impl_->drain_events() (display-deadline sweep; visualizer has no drain_events()) + ├─ artwork_->impl_->drain_events() (display-deadline sweep; visualizer has no drain_events()) + └─ source_->impl_->drain_events() (server command latch + streaming started/stopped callbacks) 5. Drain group_slot (when INBOX_TOPIC_GROUP is set in slot_bits) └─ Apply group deltas, fire on_group_update, persist last played server @@ -289,6 +303,7 @@ The `awaiting_sync_idle_events` list (on `PlayerRole::Impl`) is the key ordering - **ArtworkRole**: Stream end/clear lifecycle is handled earlier in the tick by `handle_stream_ring_event()` (dispatched from the ring drain, before this call), which clears `held_display_mask`/`display_slot` and fires `on_image_clear()` for each configured slot - preserving the "lifecycle before display" ordering the old single-function drain guaranteed. `drain_events()` itself folds any taken `display_slot` update into the main-thread-only `held_display_*` state (latest-wins per slot), then sweeps the held slots and fires `on_image_display(slot, lateness_ms)` for any whose timestamp is due on the synced client clock (or immediately if there is no active connection). The deadline is computed by the pure `display_overdue_us()` helper, which applies the slot's `display_offset_ms` shift (positive fires early) and returns the overdue microseconds; `display_lateness_ms()` maps that to the `lateness_ms` argument, reserving `0` for the no-connection case (a connected on-time display is floored to 1 ms so it never collides with that sentinel). Per-slot epochs drop a held display whose stream was replaced after the decode hand-off. `needs_drain()` ORs a nonzero `held_display_mask` into the `INBOX_TOPIC_ARTWORK_DISPLAY` bit test (the same carry-over pattern the metadata role uses for `held_delta`) so held displays keep getting a drain every tick until their deadline fires, even though the deadline sets no inbox bit; `on_image_decode` still happens on the dedicated artwork decode thread. - **Ack gate (`require_frame_done`)**: A slot can opt into per-slot back-pressure. Each `SlotBuffer` carries a `SlotAckState` (`IDLE` -> `DECODE_DELIVERED` once `on_image_decode()` fires -> `PRESENTED` once `on_image_display()`/`on_image_clear()` fires), all guarded by `slot_mutex`. While a gated slot is not `IDLE`, the decode thread (`process_notification()`) does not decode a newer notification; it *parks* it latest-wins in `SlotBuffer::parked` (`has_parked`) instead of decoding concurrently with the un-acked delivery. `ArtworkRole::frame_done(slot)` (main loop) returns the gate to `IDLE` and, if a notification is parked, calls `wake_drain_thread()` -- a sentinel `ARTWORK_RECHECK_SLOT` notification that unblocks the decode thread's `notify_queue.receive()` so it re-runs the top-of-loop parked-slot sweep (a dropped wake is covered by the `DRAIN_RECEIVE_TIMEOUT_MS` fallback). The parked notification is re-validated on replay, so a since-stale generation/epoch is simply skipped. A clear counts as a delivery: `handle_stream_ring_event()` drops any parked notification and forces gated slots to `PRESENTED`, so exactly one `frame_done()` is owed after it. A stream restart releases only `DECODE_DELIVERED` slots (their display can no longer fire); `PRESENTED` stays armed because the consumer may still be mid-fade on the prior stream's last delivery. There is no timeout. - **VisualizerRole**: Has no `drain_events()`. STREAM_START/END/CLEAR are dispatched entirely from `handle_stream_ring_event()` (from the ring drain): STREAM_START `take()`s the config from `config_slot` and fires `on_visualizer_stream_start()`; STREAM_END/CLEAR fire `on_visualizer_stream_end()`/`on_visualizer_stream_clear()`. +- **SourceRole**: Two stages. The **server command latch**: `take()` from `command_slot`, discard the command if its connection instance id no longer matches the current connection (streaming permission is per-connection), and forward only desired-state *transitions* to the task (`signal_start()`/`signal_stop()`; a start while streaming or a stop while stopped is ignored — commands are idempotent). The **stream lifecycle callbacks**: `SOURCE_STREAM` ring events land in `pending_events` during the ring drain (step 3) and are delivered here as `on_streaming_started()`/`on_streaming_stopped()`, with a `streaming_active` gate keeping the pair 1:1. The vector is indexed with a fresh `size()` check per iteration because a callback may re-enter teardown, whose `cleanup()` clears it mid-loop; `needs_drain()` ORs `!pending_events.empty()` into the `INBOX_TOPIC_SOURCE_COMMAND` bit test so same-tick events appended during the ring drain get delivered without waiting for a topic bit. ## Sync Task State Machine @@ -383,6 +398,45 @@ new_audio_client_playtime = last_finish_timestamp + remaining_buffered_frames_as This feedback loop is what makes the sync error calculation accurate. +## Source Streaming Pipeline + +The source role is the player's inverse: audio flows from the consumer's capture thread through a ring buffer to a dedicated task (`SourceTask`, `src/source_task.cpp`), which assembles it into timestamped wire chunks and sends them to the server. Like the sync task, the thread is created once and idles between streams to avoid create/destroy churn. + +### Capture ring + +`SourceRole::write_audio()` writes each call's bytes as one timestamped entry into a `SendspinAudioRingBuffer` (the same SPSC wrapper the player uses, outbound direction: producer = the consumer's capture thread, consumer = the source task). The path is non-blocking and non-allocating. An `accepting_audio_` atomic gates it: set only between `client-stream/start` being sent and the stream closing, so writes outside an open stream are rejected without touching the ring. A full ring drops the write and warns once per overflow episode (the recovery log carries the drop total); the ring's capacity (`SourceRoleConfig::capture_buffer_ms`, sized with a +1/4 metadata margin — the inverse of the player's advertise fraction) is deliberately the stall-policy backlog bound. + +### Task loop and stream binding + +The task's outer loop idles on `SOURCE_COMMAND_STOP | SOURCE_COMMAND_UPDATE`. The desired streaming state is a latest-wins atomic (`stream_requested_`) written by the role's main-loop command latch via `signal_start()`/`signal_stop()`; the task converges on it, so a start-stop-start flurry cannot be misordered by event bits. When a stream is requested, the task binds it to ONE connection for its whole life (`ConnectionManager::current_shared()`): streaming permission is per-connection (Sendspin spec, Source messages), so a drop or handoff ends the stream rather than migrating it — `stream_still_open()` re-checks `current_shared() == conn` every iteration. + +A stream then runs entirely on the task thread: + +1. **Wait for time sync** on the bound connection (chunk timestamps are meaningless before the local clock maps to the server's); same poll cadence as the sync task. +2. **Send `client-stream/start`** announcing the configured format (no negotiation; the `add_source()` config is the contract). If this send fails on a still-live connection, the task returns to idle and retries the open after a 500 ms backoff (`SOURCE_OPEN_RETRY_MS`) — the server already said start and will not repeat it, so without the retry the task would park with no wake-up coming. A stop or connection swap during the backoff is observed by the fresh desired-state and connection reads on re-entry. +3. **Flush the ring and open the gate**: the flush makes the first chunk live audio by construction, and only then is `accepting_audio_` set. A `SOURCE_STREAM` STREAMING_STARTED event is pushed onto the inbox ring. +4. **Assemble and send chunks** (below) until the desired state drops, the bound connection stops being current, or the task is told to exit. +5. **Close**: clear `accepting_audio_` first (so the tail is finite), send a final short chunk only if the encoder can take it (allowed at stream end by the spec but not required; an Opus remainder that is not a legal frame duration is dropped rather than padded), send `client-stream/end` from this same thread — ordered after the last chunk by construction — and push STREAMING_STOPPED. + +### Chunk assembly and timestamps + +Ring entries and wire chunks have independent sizes: the task copies entry bytes into a staging buffer until `chunk_duration_ms` worth of frames is assembled, consuming entries across chunk boundaries. Each ring entry's timestamp stamps its FIRST sample; when a chunk starts mid-entry, the anchor advances past the consumed frames (`source_entry_anchor_us()`). The wire timestamp is the server-domain capture time of the chunk's first sample: the local anchor minus the encoder's lookahead (so the timestamp names the audio the payload actually carries), converted with the bound connection's Kalman filter offset AND drift (`compute_server_time()`); no playback or static delay term is ever added. The staging buffer is laid out as the wire chunk itself — `[type byte 12][BE64 server-clock capture µs][payload]` — so a send is one contiguous buffer with no copy. + +### Encoder seam + +`SourceEncoder` (`src/source_encoder.h`) is the codec seam: `PcmPassthroughEncoder` returns the staged bytes untouched, and `OpusSourceEncoder` (`src/source_encoder_opus.cpp`) encodes each chunk into exactly one Opus packet (config validation guarantees one chunk is one legal Opus frame). The staging payload area is sized to `max(chunk PCM bytes, OpusSourceEncoder::MAX_PACKET_BYTES)` so no accepted config can ever drop a chunk on payload capacity. The encoder is created in `init()` and `reset()` at each stream open. + +### Stall policy + +A failed chunk send is a task-side stall: the failed chunk is dropped and the ring flushed to live (`flush_ring_to_live()`), so streaming resumes from live capture instead of bursting stale audio (Sendspin spec, Source messages — stall policy). The capture timestamps make the resulting gap self-describing to the server; the sample stream within each chunk stays continuous. Stall logging is episode-edged (one warning on entry, one recovery log with the flushed-entry count) so a stall cannot flood the log. The producer-side analogue is the full-ring drop in `write_audio()` — together the ring capacity is the "small bound" of the spec's backlog rule. + +### Outbound send path + +`SendspinConnection::send_binary_message()` is the transport contract the source task depends on (`src/connection.h`): callable from role task threads, and the completion callback fires **exactly once for every call** — inline before an error return, later from the transport, or from connection teardown for work that can never run. The task waits out every send's completion (`SOURCE_SEND_COMPLETE`) before reusing the staging buffer, which is also what makes destroying the task safe; the exactly-once contract is what keeps that unbounded wait from wedging. + +- **Host (IXWebSocket)** and **ESP client (esp_websocket_client)**: sends are synchronous in the calling thread; the callback is invoked inline. +- **ESP server (httpd)**: sends go through a per-connection **single-in-flight send slot**, allocation-free in steady state (it runs per audio chunk). The payload is copied into a connection-owned buffer sized by the first send and grown only for a larger payload; the queued worker is identified through a once-per-connection `BinarySendLookup` block holding a `weak_ptr` to the connection. If the previous send has not completed, the call returns `SsErr::NOT_FINISHED` immediately and the caller drops the chunk (the source task treats this like any failed send: a stall). The slot is released only by the worker's completion path or by the connection's destructor — never by a caller-side timeout. The worker `lock()`s the `weak_ptr` before touching anything: a connection torn down with work queued makes the worker a clean no-op (the destructor already failed the pending completion), and a locked `shared_ptr` blocks destruction until the worker returns. Binary frames are gated behind `client_hello_sent_` exactly like text frames; no binary message may legitimately precede the hello. + ## Time Synchronization ### Burst Strategy (`src/time_burst.h`) diff --git a/examples/source_client/CMakeLists.txt b/examples/source_client/CMakeLists.txt new file mode 100644 index 0000000..7206bad --- /dev/null +++ b/examples/source_client/CMakeLists.txt @@ -0,0 +1,40 @@ +# PortAudio is required: this example captures the default input device, so without +# PortAudio there is nothing to stream and the target is skipped entirely. +find_package(PkgConfig QUIET) +if(PkgConfig_FOUND) + pkg_check_modules(PORTAUDIO portaudio-2.0) +endif() + +if(NOT PORTAUDIO_FOUND) + message(STATUS "Skipping source_client example (PortAudio not found). Install portaudio " + "(brew install portaudio / apt install portaudio19-dev) to enable it.") + return() +endif() + +add_executable(source_client main.cpp) +target_link_libraries(source_client PRIVATE sendspin) +target_compile_features(source_client PRIVATE cxx_std_20) + +target_include_directories(source_client PRIVATE ${PORTAUDIO_INCLUDE_DIRS}) +target_link_directories(source_client PRIVATE ${PORTAUDIO_LIBRARY_DIRS}) +target_link_libraries(source_client PRIVATE ${PORTAUDIO_LINK_LIBRARIES}) + +# mDNS service advertisement via dns_sd.h (optional) +# macOS: built-in (no extra link flags needed — dns_sd is in libSystem) +# Linux: requires libavahi-compat-libdnssd-dev (provides dns_sd.h + libdns_sd) +if(APPLE) + target_compile_definitions(source_client PRIVATE SENDSPIN_HAS_MDNS=1) + message(STATUS "mDNS advertisement enabled (Bonjour built-in)") +else() + find_path(SOURCE_CLIENT_DNS_SD_INCLUDE_DIR dns_sd.h) + find_library(SOURCE_CLIENT_DNS_SD_LIBRARY dns_sd) + if(SOURCE_CLIENT_DNS_SD_INCLUDE_DIR AND SOURCE_CLIENT_DNS_SD_LIBRARY) + target_include_directories(source_client PRIVATE ${SOURCE_CLIENT_DNS_SD_INCLUDE_DIR}) + target_link_libraries(source_client PRIVATE ${SOURCE_CLIENT_DNS_SD_LIBRARY}) + target_compile_definitions(source_client PRIVATE SENDSPIN_HAS_MDNS=1) + message(STATUS "mDNS advertisement enabled (${SOURCE_CLIENT_DNS_SD_LIBRARY})") + else() + message(STATUS "mDNS advertisement disabled. Install libavahi-compat-libdnssd-dev " + "(Debian/Ubuntu) to enable; clients can still connect via -u ws://...") + endif() +endif() diff --git a/examples/source_client/README.md b/examples/source_client/README.md new file mode 100644 index 0000000..4e08317 --- /dev/null +++ b/examples/source_client/README.md @@ -0,0 +1,48 @@ +# Source Client Example + +Runs the sendspin-cpp client with the source role on a host computer (macOS/Linux), capturing the default PortAudio input device and streaming it to the Sendspin server. Streaming is server-gated: the client advertises the source role and waits; capture starts when the server commands the stream to start and stops when it commands stop. When built with mDNS support, advertises via mDNS so Sendspin servers discover and connect automatically; otherwise connect to a server manually with `-u ws://:/`. + +## Build + +From the repository root: + +```sh +cmake -B build +cmake --build build +``` + +The binary is at `build/examples/source_client/source_client`. + +PortAudio is required (the example is skipped without it): + +```sh +brew install portaudio # macOS +sudo apt install portaudio19-dev # Debian/Ubuntu +``` + +The example builds only when the source role is enabled (`SENDSPIN_ENABLE_SOURCE`, on by default). + +### Linux prerequisites + +mDNS service advertisement is optional. Install Avahi's Bonjour-compatible headers to enable it: + +```sh +sudo apt install libavahi-compat-libdnssd-dev +``` + +macOS has mDNS support built in; no extra dependencies needed. + +## Run + +```sh +./build/examples/source_client/source_client # default name "Source Client" +./build/examples/source_client/source_client "Line In" # custom name +./build/examples/source_client/source_client -o # stream Opus instead of PCM +./build/examples/source_client/source_client -p 8930 # listen on a custom port +``` + +The client listens on port 8928 by default. When mDNS is enabled it advertises `_sendspin._tcp` with the configured port so Sendspin servers on the local network discover and connect automatically; otherwise tell the server to connect with `ws://:8928/sendspin`, replacing `8928` if you passed `-p`. + +Audio is captured at 48 kHz / 16-bit from the default input device (mono or stereo, following the device). By default chunks are sent as raw PCM; with `-o` each 20 ms chunk is Opus-encoded before sending. + +Press Ctrl+C to stop. diff --git a/examples/source_client/main.cpp b/examples/source_client/main.cpp new file mode 100644 index 0000000..90c8514 --- /dev/null +++ b/examples/source_client/main.cpp @@ -0,0 +1,484 @@ +// 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 Host example application streaming audio capture to a Sendspin server. +/// +/// Runs a SendspinClient with the source role on the host computer, capturing +/// the default PortAudio input device and streaming it to the server when the +/// server commands the stream to start. When built with mDNS support +/// (dns_sd.h available), advertises via mDNS so Sendspin servers can discover +/// and connect automatically; otherwise the user must connect manually with +/// `-u ws://:/`. +/// +/// Usage: ./source_client [options] [name] +/// name: Optional friendly name (default: "Source Client") +/// +/// Options: +/// -u URL Connect to a WebSocket URL (e.g. ws://192.168.1.10:8928/sendspin) +/// -p PORT Listen on PORT (default: 8928) +/// -o Stream Opus-encoded audio instead of PCM +/// -l LEVEL Set log level: none, error, warn, info (default), debug, verbose +/// -v Verbose logging (same as -l verbose) +/// -q Quiet logging (same as -l error) +/// -h Show usage + +#include "sendspin/client.h" +#include "sendspin/source_role.h" + +#include +#include + +#ifdef SENDSPIN_HAS_MDNS +#include +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace sendspin; + +static constexpr uint16_t DEFAULT_SENDSPIN_PORT = SendspinClientConfig::DEFAULT_SERVER_PORT; +static const char* SENDSPIN_PATH = "/sendspin"; + +// 48 kHz / 16-bit: a format PortAudio input backends deliver everywhere, and one of the +// sample rates libopus accepts natively, so the same capture format is legal for both the +// PCM default and the -o Opus path. The channel count follows the default input device +// (clamped to stereo) since many capture devices are mono. +static constexpr uint32_t CAPTURE_SAMPLE_RATE = 48000; +static constexpr uint8_t CAPTURE_BIT_DEPTH = 16; + +// One outbound chunk must be exactly one legal Opus frame; 20 ms is Opus's canonical frame +// duration, splitting the difference between capture latency (10 ms) and per-packet +// overhead (40/60 ms). The PCM path keeps the library's default chunk duration. +static constexpr uint32_t OPUS_CHUNK_MS = 20; + +#ifdef SENDSPIN_HAS_MDNS +// Manages mDNS service advertisement via dns_sd.h +class MdnsAdvertiser { +public: + ~MdnsAdvertiser() { + stop(); + } + + bool start(const std::string& name, uint16_t port, const std::string& path) { + // Build TXT record with path and name keys + TXTRecordRef txt; + TXTRecordCreate(&txt, 0, nullptr); + TXTRecordSetValue(&txt, "path", static_cast(path.size()), path.c_str()); + TXTRecordSetValue(&txt, "name", static_cast(name.size()), name.c_str()); + + DNSServiceErrorType err = DNSServiceRegister( + &service_ref_, + 0, // flags + 0, // interface index (0 = all) + name.c_str(), // service name + "_sendspin._tcp", // service type + nullptr, // domain (default) + nullptr, // host (default) + htons(port), // port (network byte order) + TXTRecordGetLength(&txt), + TXTRecordGetBytesPtr(&txt), + nullptr, // callback (not needed for simple registration) + nullptr // context + ); + + TXTRecordDeallocate(&txt); + + if (err != kDNSServiceErr_NoError) { + fprintf(stderr, "Failed to register mDNS service: error %d\n", err); + return false; + } + + fprintf(stderr, "mDNS: Advertising _sendspin._tcp on port %u (name: %s)\n", port, + name.c_str()); + return true; + } + + void stop() { + if (service_ref_ != nullptr) { + DNSServiceRefDeallocate(service_ref_); + service_ref_ = nullptr; + fprintf(stderr, "mDNS: Service advertisement stopped\n"); + } + } + +private: + DNSServiceRef service_ref_{nullptr}; +}; +#endif // SENDSPIN_HAS_MDNS + +/// @brief Captures the default PortAudio input device and feeds the source role: the input +/// callback forwards each buffer to write_audio() (callback-safe) with its capture time mapped +/// onto the local steady clock; the listener callbacks start and stop the capture stream. +class PortAudioCapture { +public: + PortAudioCapture() { + PaError err = Pa_Initialize(); + initialized_ = (err == paNoError); + if (!initialized_) { + fprintf(stderr, "Pa_Initialize failed: %s\n", Pa_GetErrorText(err)); + } + } + + ~PortAudioCapture() { + stop(); + if (stream_ != nullptr) { + Pa_CloseStream(stream_); + } + if (initialized_) { + Pa_Terminate(); + } + } + + // Not copyable or movable (the PortAudio stream holds a pointer to this) + PortAudioCapture(const PortAudioCapture&) = delete; + PortAudioCapture& operator=(const PortAudioCapture&) = delete; + + /// @brief Returns the default input device's channel count clamped to stereo, or 0 when + /// no input device is available. + uint8_t default_input_channels() const { + if (!initialized_) { + return 0; + } + PaDeviceIndex device = Pa_GetDefaultInputDevice(); + if (device == paNoDevice) { + return 0; + } + const PaDeviceInfo* info = Pa_GetDeviceInfo(device); + if (info == nullptr || info->maxInputChannels <= 0) { + return 0; + } + return static_cast(std::min(info->maxInputChannels, 2)); + } + + /// @brief Opens (but does not start) the capture stream on the default input device. + bool open(SourceRole& source, uint32_t sample_rate, uint8_t channels) { + if (!initialized_) { + return false; + } + source_ = &source; + bytes_per_frame_ = static_cast(channels) * (CAPTURE_BIT_DEPTH / 8U); + + PaStreamParameters params; + memset(¶ms, 0, sizeof(params)); + params.device = Pa_GetDefaultInputDevice(); + if (params.device == paNoDevice) { + fprintf(stderr, "No default input device available\n"); + return false; + } + params.channelCount = channels; + params.sampleFormat = paInt16; + params.suggestedLatency = Pa_GetDeviceInfo(params.device)->defaultLowInputLatency; + + PaError err = Pa_OpenStream(&stream_, ¶ms, nullptr, sample_rate, + paFramesPerBufferUnspecified, paClipOff, pa_callback, this); + if (err != paNoError) { + fprintf(stderr, "Pa_OpenStream failed: %s\n", Pa_GetErrorText(err)); + stream_ = nullptr; + return false; + } + const PaDeviceInfo* info = Pa_GetDeviceInfo(params.device); + fprintf(stderr, "Capturing from \"%s\" (%u Hz, %u ch)\n", info->name, sample_rate, + channels); + return true; + } + + /// @brief Starts capture; the callback begins feeding write_audio(). + bool start() { + if (stream_ == nullptr || Pa_IsStreamActive(stream_) == 1) { + return stream_ != nullptr; + } + PaError err = Pa_StartStream(stream_); + if (err != paNoError) { + fprintf(stderr, "Pa_StartStream failed: %s\n", Pa_GetErrorText(err)); + return false; + } + return true; + } + + /// @brief Stops capture, draining the callback before returning. + void stop() { + if (stream_ != nullptr && Pa_IsStreamActive(stream_) == 1) { + Pa_StopStream(stream_); + } + } + + /// @brief Returns the number of rejected writes since the last call and resets the count. + uint32_t take_dropped_writes() { + return dropped_writes_.exchange(0, std::memory_order_relaxed); + } + +private: + static int pa_callback(const void* input, void* /*output*/, unsigned long frame_count, + const PaStreamCallbackTimeInfo* time_info, + PaStreamCallbackFlags /*status_flags*/, void* user_data) { + auto* self = static_cast(user_data); + if (input == nullptr) { + return paContinue; // Input overflow gap; nothing to forward + } + + // Only the buffer's age (currentTime - inputBufferAdcTime) is portable across + // PortAudio backends; subtract it from the local steady clock to get the capture time + // in the client's domain. Backends reporting zero timestamps get the library's + // documented pass-0 arrival-stamp fallback. + int64_t capture_us = 0; + if (time_info != nullptr && time_info->currentTime > 0 && + time_info->inputBufferAdcTime > 0) { + int64_t now_us = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + double age_us = + (time_info->currentTime - time_info->inputBufferAdcTime) * 1000000.0; + capture_us = now_us - static_cast(std::round(age_us)); + } + + if (!self->source_->write_audio(static_cast(input), + frame_count * self->bytes_per_frame_, capture_us)) { + // Counted here and reported from the main loop: no logging on the audio + // callback. A rejected write means the capture ring is full, or the stream + // closed while this callback was in flight. + self->dropped_writes_.fetch_add(1, std::memory_order_relaxed); + } + return paContinue; + } + + SourceRole* source_{nullptr}; + PaStream* stream_{nullptr}; + size_t bytes_per_frame_{0}; + std::atomic dropped_writes_{0}; + bool initialized_{false}; +}; + +static std::atomic running{true}; + +static void signal_handler(int /*sig*/) { + running.store(false); +} + +static void print_usage(const char* prog) { + fprintf(stderr, "Usage: %s [options] [name]\n", prog); + fprintf(stderr, " name Friendly name (default: \"Source Client\")\n\n"); + fprintf(stderr, "Options:\n"); + fprintf(stderr, " -u URL Connect to a WebSocket URL (e.g. ws://192.168.1.10:8928/sendspin)\n"); + fprintf(stderr, " -p PORT Listen on PORT (default: %u)\n", DEFAULT_SENDSPIN_PORT); + fprintf(stderr, " -o Stream Opus-encoded audio instead of PCM\n"); + fprintf(stderr, " -l LEVEL Log level: none, error, warn, info (default), debug, verbose\n"); + fprintf(stderr, " -v Verbose logging (same as -l verbose)\n"); + fprintf(stderr, " -q Quiet logging (same as -l error)\n"); + fprintf(stderr, " -h Show this help\n"); +} + +static bool parse_log_level(const char* str, LogLevel& level) { + if (strcmp(str, "none") == 0) { level = LogLevel::NONE; return true; } + if (strcmp(str, "error") == 0) { level = LogLevel::ERROR; return true; } + if (strcmp(str, "warn") == 0) { level = LogLevel::WARN; return true; } + if (strcmp(str, "info") == 0) { level = LogLevel::INFO; return true; } + if (strcmp(str, "debug") == 0) { level = LogLevel::DEBUG; return true; } + if (strcmp(str, "verbose") == 0) { level = LogLevel::VERBOSE; return true; } + return false; +} + +static bool parse_port(const char* str, uint16_t& port) { + char* end = nullptr; + unsigned long value = strtoul(str, &end, 10); + if (*str == '\0' || *end != '\0' || value == 0 || value > 65535UL) { + return false; + } + port = static_cast(value); + return true; +} + +int main(int argc, char* argv[]) { + // Set up signal handler for clean shutdown + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); + + // Parse command line options + LogLevel log_level = LogLevel::INFO; + std::string connect_url; + uint16_t server_port = DEFAULT_SENDSPIN_PORT; + bool use_opus = false; + int opt; + while ((opt = getopt(argc, argv, "u:p:ol:vqh")) != -1) { + switch (opt) { + case 'u': + connect_url = optarg; + break; + case 'p': + if (!parse_port(optarg, server_port)) { + fprintf(stderr, "Invalid port: %s\n", optarg); + print_usage(argv[0]); + return 1; + } + break; + case 'o': + use_opus = true; + break; + case 'l': + if (!parse_log_level(optarg, log_level)) { + fprintf(stderr, "Unknown log level: %s\n", optarg); + print_usage(argv[0]); + return 1; + } + break; + case 'v': + log_level = LogLevel::VERBOSE; + break; + case 'q': + log_level = LogLevel::ERROR; + break; + case 'h': + print_usage(argv[0]); + return 0; + default: + print_usage(argv[0]); + return 1; + } + } + + SendspinClient::set_log_level(log_level); + + // Optional name from remaining arguments + std::string friendly_name = (optind < argc) ? argv[optind] : "Source Client"; + + // The capture channel count comes from the hardware, and the format is fixed at + // add_source() time, so probe the device before configuring the client. + PortAudioCapture capture; + uint8_t channels = capture.default_input_channels(); + if (channels == 0) { + fprintf(stderr, "No usable input device; cannot stream\n"); + return 1; + } + + // Configure the client + SendspinClientConfig config; + config.client_id = "source-client-example"; + config.name = friendly_name; + config.product_name = "sendspin-cpp host example"; + config.manufacturer = "sendspin-cpp"; + config.software_version = "0.1.0"; + config.server_port = server_port; + + SendspinClient client(std::move(config)); + + // Add the source role. The config is the capture format contract for every stream this + // role opens; Opus narrows it to one legal frame duration per chunk. + SourceRoleConfig source_config{ + .sample_rate = CAPTURE_SAMPLE_RATE, + .chunk_duration_ms = + use_opus ? OPUS_CHUNK_MS : SourceRoleConfig::DEFAULT_CHUNK_MS, + .codec = use_opus ? SendspinCodecFormat::OPUS : SendspinCodecFormat::PCM, + .channels = channels, + .bit_depth = CAPTURE_BIT_DEPTH, + }; + auto& source = client.add_source(source_config); + + if (!capture.open(source, CAPTURE_SAMPLE_RATE, channels)) { + return 1; + } + + // --- Listener implementations --- + + struct CaptureSourceListener : SourceRoleListener { + PortAudioCapture& capture; + explicit CaptureSourceListener(PortAudioCapture& c) : capture(c) {} + + void on_streaming_started() override { + fprintf(stderr, ">>> Streaming started\n"); + if (!capture.start()) { + fprintf(stderr, ">>> Failed to start capture\n"); + } + } + + void on_streaming_stopped() override { + fprintf(stderr, ">>> Streaming stopped\n"); + capture.stop(); + } + }; + + struct HostNetworkProvider : SendspinNetworkProvider { + bool is_network_ready() override { return true; } + }; + + CaptureSourceListener source_listener(capture); + HostNetworkProvider network_provider; + + source.set_listener(&source_listener); + client.set_network_provider(&network_provider); + + // Start the server + fprintf(stderr, "Starting Sendspin source client on port %u (%s)...\n", server_port, + use_opus ? "opus" : "pcm"); + + if (!client.start_server()) { + fprintf(stderr, "Failed to start server\n"); + return 1; + } + +#ifdef SENDSPIN_HAS_MDNS + MdnsAdvertiser mdns; + if (!mdns.start(friendly_name, server_port, SENDSPIN_PATH)) { + fprintf(stderr, "Warning: mDNS advertisement failed, server still running\n"); + fprintf(stderr, "Connect manually to ws://:%u%s\n", server_port, + SENDSPIN_PATH); + } +#else + fprintf(stderr, + "mDNS advertisement not compiled in. Either restart with " + "-u ws://:/ to dial a server, or tell a server " + "to connect to ws://:%u%s.\n", + server_port, SENDSPIN_PATH); +#endif + + // Auto-connect if a URL was provided via -u + if (!connect_url.empty()) { + fprintf(stderr, "Connecting to %s...\n", connect_url.c_str()); + client.connect_to(connect_url); + } + + fprintf(stderr, "Press Ctrl+C to stop. The server controls when streaming starts.\n\n"); + + // Main loop + int tick = 0; + while (running.load()) { + client.loop(); + // Surface capture drops off the audio callback (which only counts them) + if (++tick % 500 == 0) { + uint32_t dropped = capture.take_dropped_writes(); + if (dropped > 0) { + fprintf(stderr, ">>> Dropped %u capture writes in the last 5 s\n", dropped); + } + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + fprintf(stderr, "\nShutting down...\n"); +#ifdef SENDSPIN_HAS_MDNS + mdns.stop(); +#endif + // Stop feeding audio before the client tears down the stream + capture.stop(); + client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + + return 0; +} From 842d16a20da156589dcc06630b4284913818a1c6 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Fri, 4 Sep 2026 18:14:02 -0500 Subject: [PATCH 2/2] Address the example and documentation review round The example declares its listener and network provider before the client (their raw pointers must outlive it), exits the main loop when capture fails to start so shutdown closes the open stream, reports capture drops on a steady-clock interval instead of counted loop ticks, warns at startup that any handshaken server can start capture on this protocol revision, and corrects the audio-callback logging comment. The integration guide gains the matching authorization note and the 20 ms default with 5 ms Opus support; internals documents the binary send worker, its slot, and its reclamation path. --- docs/integration-guide.md | 4 +- docs/internals.md | 2 +- examples/source_client/main.cpp | 79 +++++++++++++++++++-------------- 3 files changed, 50 insertions(+), 35 deletions(-) diff --git a/docs/integration-guide.md b/docs/integration-guide.md index a7a13a6..81d0c97 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -188,6 +188,8 @@ The configuration is validated at `add_source()` time; validation fails closed. Streaming is gated by the server: the client never streams unsolicited, the default after connect is stopped, and permission does not survive reconnection. When the server commands start, the role opens the outbound stream and fires `on_streaming_started()`; from that point on, feed captured audio to `write_audio()`: +> **Authorization on this protocol revision:** the spec additionally gates source activation on a paired (`user`-trust) connection, with the decision on the server side. This library predates the pairing/encryption transport, so the start command of any server that completed the handshake is honored; treat every network the client can be reached from as trusted until the encryption work lands, and prefer wired/isolated networks for privacy-sensitive inputs such as microphones. + ```cpp // Capture thread (exactly one producer thread): source.write_audio(pcm_bytes, len, capture_time_us); @@ -937,7 +939,7 @@ Configuration passed to `client.add_source()`. It is the capture format contract | `sample_rate` | `uint32_t` | `48000` | Capture sample rate in Hz; must be > 0. `OPUS` accepts only libopus's rates: 8000, 12000, 16000, 24000, or 48000 (a 44100 line-in must use `PCM` or resample upstream). | | `channels` | `uint8_t` | `2` | Capture channel count; must be > 0 (1 or 2 for `OPUS`). | | `bit_depth` | `uint8_t` | `16` | Bits per sample: 16, 24 (3 packed bytes per sample), or 32. `OPUS` requires 16. | -| `chunk_duration_ms` | `uint32_t` | `25` | Duration of one outbound audio chunk. `PCM` accepts the spec bounds [5, 150]; `OPUS` accepts only 10, 20, 40, or 60 (one chunk is exactly one legal Opus frame), so the PCM default of 25 is rejected for `OPUS`. | +| `chunk_duration_ms` | `uint32_t` | `20` | Duration of one outbound audio chunk. `PCM` accepts the spec bounds [5, 150]; `OPUS` accepts only 5, 10, 20, 40, or 60 (one chunk is exactly one legal Opus frame). The default is a legal Opus frame, so switching `codec` alone keeps a valid config. | | `capture_buffer_ms` | `uint32_t` | `150` | Capture ring capacity in milliseconds of audio in the configured format; must be > 0. The ring is the stall-policy backlog bound: capture beyond it is dropped at `write_audio()` and streaming resumes from live audio. Approximate: per-write ring metadata comes out of a fixed +25% margin, so many very small `write_audio()` calls reduce the effective audio capacity below this figure. | | `opus_bitrate` | `uint32_t` | `128000` | Opus bitrate in bit/s, validated against [500, 512000] (the range libopus accepts). Ignored (and unvalidated) when `codec` is `PCM`. | | `opus_complexity` | `uint8_t` | `2` | Opus encoder complexity, validated to at most 10. The low default fits an ESP32-class real-time encode budget; hosts may raise it for quality per CPU. Ignored (and unvalidated) when `codec` is `PCM`. | diff --git a/docs/internals.md b/docs/internals.md index 167a77e..e655a43 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -516,7 +516,7 @@ On the ESP build, `SendspinServerConnection` lifetime is pinned to the httpd ses 3. The httpd WebSocket handler (`websocket_handler`) looks the connection up by `httpd_sess_get_ctx(handle, sockfd)` at run time, copying the slot's `shared_ptr` for the duration of its work; it never assumes the manager's observer slot is alive. The queued send workers (`async_send_text`, `async_send_time_text`) instead capture a `weak_ptr` to the originating connection and `lock()` it when they run. 4. When the socket closes, httpd calls the `close_fn` first (which fires `connection_closed_callback_` so `ConnectionManager` can drop its observer in the next `loop()`), then later calls the slot's `free_fn` to release the authoritative reference once no workers are queued for that session. -Queued send workers capture a `weak_ptr` to the originating connection — `AsyncRespArg` for text sends, `SessionLookup` for time sends — and `lock()` it when the worker runs. This is deliberately **not** a `{httpd_handle_t, int sockfd}` pair: identifying the target by sockfd risked binding to a *different* connection that had recycled the same fd after the original closed, sending a frame to the wrong peer. The `weak_ptr` resolves to the exact connection that queued the work, or to null if it has since been destroyed, in which case the worker no-ops cleanly. Because these structs now hold non-trivial members (the `weak_ptr`, and `AsyncRespArg`'s completion `std::function`), they are constructed with placement-new and explicitly destroyed before `platform_free` rather than treated as POD. Both are allocated through `platform_malloc` / `platform_malloc_internal`. +Queued send workers capture a `weak_ptr` to the originating connection — `AsyncRespArg` for text sends, `SessionLookup` for time sends, `BinarySendLookup` for binary sends — and `lock()` it when the worker runs. Unlike the per-message text contexts, `BinarySendLookup` is a reusable once-per-connection block backing the single-in-flight binary send slot: `async_send_binary` sends from the connection-owned payload buffer, releases the slot, and fires the completion exactly once on every exit path; the slot is otherwise released only by the connection destructor, and blocks whose queued worker was discarded by `httpd_stop` are reclaimed by `reclaim_orphaned_binary_send_work()` from the ws server's stop path. This is deliberately **not** a `{httpd_handle_t, int sockfd}` pair: identifying the target by sockfd risked binding to a *different* connection that had recycled the same fd after the original closed, sending a frame to the wrong peer. The `weak_ptr` resolves to the exact connection that queued the work, or to null if it has since been destroyed, in which case the worker no-ops cleanly. Because these structs now hold non-trivial members (the `weak_ptr`, and `AsyncRespArg`'s completion `std::function`), they are constructed with placement-new and explicitly destroyed before `platform_free` rather than treated as POD. Both are allocated through `platform_malloc` / `platform_malloc_internal`. 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. diff --git a/examples/source_client/main.cpp b/examples/source_client/main.cpp index 90c8514..9f2eca3 100644 --- a/examples/source_client/main.cpp +++ b/examples/source_client/main.cpp @@ -254,9 +254,10 @@ class PortAudioCapture { if (!self->source_->write_audio(static_cast(input), frame_count * self->bytes_per_frame_, capture_us)) { - // Counted here and reported from the main loop: no logging on the audio - // callback. A rejected write means the capture ring is full, or the stream - // closed while this callback was in flight. + // Counted here and reported from the main loop; this example adds no logging of + // its own on the audio callback (the library itself warns once per overflow + // episode from this thread). A rejected write means the capture ring is full, or + // the stream closed while this callback was in flight. self->dropped_writes_.fetch_add(1, std::memory_order_relaxed); } return paContinue; @@ -370,6 +371,37 @@ int main(int argc, char* argv[]) { return 1; } + // --- Listener implementations --- + // Declared before the client: the client and role retain raw pointers to these for their + // whole lifetime, so they must be destroyed after the client (reverse declaration order). + + struct CaptureSourceListener : SourceRoleListener { + PortAudioCapture& capture; + explicit CaptureSourceListener(PortAudioCapture& c) : capture(c) {} + + void on_streaming_started() override { + fprintf(stderr, ">>> Streaming started\n"); + if (!capture.start()) { + // The stream is open on the wire but capture cannot run; exit the main loop so + // shutdown closes the stream instead of leaving the server waiting on silence. + fprintf(stderr, ">>> Failed to start capture; shutting down\n"); + running.store(false); + } + } + + void on_streaming_stopped() override { + fprintf(stderr, ">>> Streaming stopped\n"); + capture.stop(); + } + }; + + struct HostNetworkProvider : SendspinNetworkProvider { + bool is_network_ready() override { return true; } + }; + + CaptureSourceListener source_listener(capture); + HostNetworkProvider network_provider; + // Configure the client SendspinClientConfig config; config.client_id = "source-client-example"; @@ -397,32 +429,6 @@ int main(int argc, char* argv[]) { return 1; } - // --- Listener implementations --- - - struct CaptureSourceListener : SourceRoleListener { - PortAudioCapture& capture; - explicit CaptureSourceListener(PortAudioCapture& c) : capture(c) {} - - void on_streaming_started() override { - fprintf(stderr, ">>> Streaming started\n"); - if (!capture.start()) { - fprintf(stderr, ">>> Failed to start capture\n"); - } - } - - void on_streaming_stopped() override { - fprintf(stderr, ">>> Streaming stopped\n"); - capture.stop(); - } - }; - - struct HostNetworkProvider : SendspinNetworkProvider { - bool is_network_ready() override { return true; } - }; - - CaptureSourceListener source_listener(capture); - HostNetworkProvider network_provider; - source.set_listener(&source_listener); client.set_network_provider(&network_provider); @@ -456,17 +462,24 @@ int main(int argc, char* argv[]) { client.connect_to(connect_url); } - fprintf(stderr, "Press Ctrl+C to stop. The server controls when streaming starts.\n\n"); + fprintf(stderr, "Press Ctrl+C to stop. The server controls when streaming starts.\n"); + fprintf(stderr, + "NOTE: on this protocol revision any Sendspin server that completes the handshake\n" + "can start capture from the default input; keep this example on trusted networks\n" + "(pairing-based authorization arrives with the encryption work).\n\n"); // Main loop - int tick = 0; + constexpr auto DROP_REPORT_INTERVAL = std::chrono::seconds(5); + auto next_drop_report = std::chrono::steady_clock::now() + DROP_REPORT_INTERVAL; while (running.load()) { client.loop(); // Surface capture drops off the audio callback (which only counts them) - if (++tick % 500 == 0) { + if (std::chrono::steady_clock::now() >= next_drop_report) { + next_drop_report += DROP_REPORT_INTERVAL; uint32_t dropped = capture.take_dropped_writes(); if (dropped > 0) { - fprintf(stderr, ">>> Dropped %u capture writes in the last 5 s\n", dropped); + fprintf(stderr, ">>> Dropped %u capture writes in the last %lld s\n", dropped, + static_cast(DROP_REPORT_INTERVAL.count())); } } std::this_thread::sleep_for(std::chrono::milliseconds(10));