diff --git a/CMakeLists.txt b/CMakeLists.txt index c235bea..c35af57 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -78,6 +78,16 @@ if(ESP_IDF_BUILD) list(APPEND SENDSPIN_COMPILE_DEFS SENDSPIN_ENABLE_VISUALIZER) endif() + if(CONFIG_SENDSPIN_ENABLE_SOURCE) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_SOURCE_SOURCES}) + list(APPEND SENDSPIN_COMPILE_DEFS SENDSPIN_ENABLE_SOURCE) + endif() + + # Shared audio infrastructure for the roles that stream audio through the SPSC ring + if(CONFIG_SENDSPIN_ENABLE_PLAYER OR CONFIG_SENDSPIN_ENABLE_SOURCE) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_AUDIO_SOURCES}) + endif() + # Register the component — uses assembled source list idf_component_register( SRCS ${SENDSPIN_ALL_SOURCES} @@ -109,6 +119,7 @@ else() option(SENDSPIN_ENABLE_COLOR "Enable color role" ON) option(SENDSPIN_ENABLE_ARTWORK "Enable artwork role" ON) option(SENDSPIN_ENABLE_VISUALIZER "Enable visualizer role" ON) + option(SENDSPIN_ENABLE_SOURCE "Enable source role (audio capture)" ON) include(${CMAKE_CURRENT_SOURCE_DIR}/cmake/host.cmake) @@ -136,6 +147,13 @@ else() if(SENDSPIN_ENABLE_VISUALIZER) list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_VISUALIZER_SOURCES}) endif() + if(SENDSPIN_ENABLE_SOURCE) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_SOURCE_SOURCES}) + endif() + # Shared audio infrastructure for the roles that stream audio through the SPSC ring + if(SENDSPIN_ENABLE_PLAYER OR SENDSPIN_ENABLE_SOURCE) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_AUDIO_SOURCES}) + endif() # Create the library — core sources + enabled role sources + host networking (added via host.cmake) add_library(sendspin STATIC ${SENDSPIN_ALL_SOURCES}) @@ -159,6 +177,9 @@ else() if(SENDSPIN_ENABLE_VISUALIZER) target_compile_definitions(sendspin PUBLIC SENDSPIN_ENABLE_VISUALIZER) endif() + if(SENDSPIN_ENABLE_SOURCE) + target_compile_definitions(sendspin PUBLIC SENDSPIN_ENABLE_SOURCE) + endif() # Host build options option(ENABLE_WERROR "Treat warnings as errors" OFF) diff --git a/Kconfig b/Kconfig index 6c72468..a69c82b 100644 --- a/Kconfig +++ b/Kconfig @@ -35,4 +35,8 @@ menu "sendspin-cpp" bool "Enable visualizer role" default y + config SENDSPIN_ENABLE_SOURCE + bool "Enable source role (audio capture)" + default y + endmenu diff --git a/cmake/sources.cmake b/cmake/sources.cmake index d32f30c..5ef0b5e 100644 --- a/cmake/sources.cmake +++ b/cmake/sources.cmake @@ -26,16 +26,30 @@ function(sendspin_get_sources BASE_DIR) PARENT_SCOPE ) + # Shared audio sources — appended when any role that moves audio through the SPSC ring + # buffer is enabled (player or source), so the filename lives in exactly one list + set(SENDSPIN_AUDIO_SOURCES + ${BASE_DIR}/src/audio_ring_buffer.cpp + + PARENT_SCOPE + ) + # Per-role source sets — conditionally compiled based on SENDSPIN_ENABLE_* options set(SENDSPIN_PLAYER_SOURCES ${BASE_DIR}/src/player_role.cpp - ${BASE_DIR}/src/audio_ring_buffer.cpp ${BASE_DIR}/src/decoder.cpp ${BASE_DIR}/src/sync_task.cpp PARENT_SCOPE ) + set(SENDSPIN_SOURCE_SOURCES + ${BASE_DIR}/src/source_role.cpp + ${BASE_DIR}/src/source_task.cpp + + PARENT_SCOPE + ) + set(SENDSPIN_CONTROLLER_SOURCES ${BASE_DIR}/src/controller_role.cpp diff --git a/include/sendspin/client.h b/include/sendspin/client.h index 5f5fdf2..6677c79 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -47,6 +47,9 @@ class MetadataRole; #ifdef SENDSPIN_ENABLE_PLAYER class PlayerRole; #endif +#ifdef SENDSPIN_ENABLE_SOURCE +class SourceRole; +#endif #ifdef SENDSPIN_ENABLE_VISUALIZER class VisualizerRole; #endif @@ -258,6 +261,11 @@ class SendspinClient { VisualizerRole& add_visualizer(VisualizerRoleConfig config); #endif +#ifdef SENDSPIN_ENABLE_SOURCE + /// @brief Adds the source role. Returns a reference for setting callbacks + SourceRole& add_source(SourceRoleConfig config); +#endif + // ======================================== // Role access (nullptr if not added) // ======================================== @@ -322,6 +330,18 @@ class SendspinClient { return this->player_.get(); } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + /// @brief Returns the source role, or nullptr if not added + /// @return Pointer to the source role, or nullptr + SourceRole* source() { + return this->source_.get(); + } + /// @brief Returns the source role (const), or nullptr if not added + /// @return Const pointer to the source role, or nullptr + const SourceRole* source() const { + return this->source_.get(); + } +#endif #ifdef SENDSPIN_ENABLE_VISUALIZER /// @brief Returns the visualizer role, or nullptr if not added /// @return Pointer to the visualizer role, or nullptr @@ -490,6 +510,9 @@ class SendspinClient { SendspinPersistenceProvider* persistence_provider_{nullptr}; #ifdef SENDSPIN_ENABLE_PLAYER std::unique_ptr player_; +#endif +#ifdef SENDSPIN_ENABLE_SOURCE + std::unique_ptr source_; #endif std::unique_ptr time_burst_; #ifdef SENDSPIN_ENABLE_VISUALIZER diff --git a/include/sendspin/config.h b/include/sendspin/config.h index 89977c6..b110091 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -107,7 +107,7 @@ struct SendspinClientConfig { // Player config types // ============================================================================ -/// @brief Audio codec format for a player stream +/// @brief Audio codec format for an audio stream (player playback or source capture) enum class SendspinCodecFormat : uint8_t { FLAC, // FLAC lossless audio OPUS, // Opus compressed audio @@ -266,4 +266,74 @@ struct VisualizerRoleConfig { unsigned priority{2}; ///< FreeRTOS priority for the drain thread (ESP-IDF only) }; +// ============================================================================ +// Source config types +// ============================================================================ + +/// @brief Configuration for the source role (audio capture streamed to the server) +/// +/// The configured format is the contract for every stream the role opens: there is no +/// negotiation, and write_audio() bytes are forwarded untouched. An invalid config leaves the +/// role added but inert (logged at ERROR; the role is not advertised and never streams) -- +/// spec-invalid values are rejected, never clamped or repaired. +struct SourceRoleConfig { + /// @brief Chunk duration bounds from the Sendspin spec (Source messages): chunks MUST be + /// at most 150 ms and SHOULD be at least 5 ms + static constexpr uint32_t CHUNK_MIN_MS = 5U; + static constexpr uint32_t CHUNK_MAX_MS = 150U; + + /// @brief Default duration (ms) of one outbound audio chunk. Small enough to keep the + /// capture-to-server latency and per-chunk staging buffer modest, large enough that the + /// per-chunk framing/send overhead stays negligible; well inside the spec bounds above, + /// and a legal Opus frame duration so switching codec alone never invalidates a default + /// config + static constexpr uint32_t DEFAULT_CHUNK_MS = 20U; + + /// @brief Default capture ring capacity in milliseconds. Derived from the spec's maximum + /// chunk duration: the ring IS the "small bound" of the spec's stall policy (Source + /// messages) -- backlog beyond it is dropped at write_audio() and streaming resumes from + /// live capture rather than bursting stale audio + static constexpr uint32_t DEFAULT_CAPTURE_BUFFER_MS = CHUNK_MAX_MS; + + /// @brief Default FreeRTOS priority for the source task (ESP-IDF only). Below the HTTP + /// server task (SendspinClientConfig::DEFAULT_HTTPD_PRIORITY = 5) and the sync/decode task + /// (6) so outbound capture can never starve inbound playback, above the artwork/visualizer + /// drain threads (2) + static constexpr unsigned DEFAULT_SOURCE_TASK_PRIORITY = 3U; + + /// @brief Default capture sample rate (Hz): the native rate of most capture hardware + static constexpr uint32_t DEFAULT_SOURCE_SAMPLE_RATE = 48000U; + + // 32-bit fields + uint32_t sample_rate{DEFAULT_SOURCE_SAMPLE_RATE}; ///< Capture sample rate in Hz; must be > 0 + + /// @brief Outbound chunk duration in milliseconds, validated against the spec bounds + /// [CHUNK_MIN_MS, CHUNK_MAX_MS] + uint32_t chunk_duration_ms{DEFAULT_CHUNK_MS}; + + /// @brief Capture ring capacity in milliseconds of audio in the configured format (the + /// byte size is computed from sample_rate, channels, and bit_depth). See + /// DEFAULT_CAPTURE_BUFFER_MS for why this doubles as the stall-policy backlog bound. + /// 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 + uint32_t capture_buffer_ms{DEFAULT_CAPTURE_BUFFER_MS}; + + unsigned priority{DEFAULT_SOURCE_TASK_PRIORITY}; ///< FreeRTOS priority for the source + ///< task (ESP-IDF only) + + /// @brief Memory placement for the capture ring and chunk staging buffer (ESP-IDF only; + /// ignored on host). Bulk audio with sequential access, so PREFER_EXTERNAL (SPIRAM) -- + /// mirrors the player decode buffer's choice + MemoryLocation buffer_location{MemoryLocation::PREFER_EXTERNAL}; + + // 8-bit fields + /// @brief Outbound codec. Only PCM is accepted; an OPUS config is rejected as inert until + /// Opus encoding lands + SendspinCodecFormat codec{SendspinCodecFormat::PCM}; + uint8_t channels{2}; ///< Capture channel count; must be > 0 + uint8_t bit_depth{16}; ///< Bits per sample; 16, 24 (3 packed bytes), or 32 + bool line_sense{false}; ///< Advertise line-input signal sensing (see SourceRole::set_signal) + bool psram_stack{false}; ///< Allocate source task stack in PSRAM (ESP-IDF only) +}; + } // namespace sendspin diff --git a/include/sendspin/source_role.h b/include/sendspin/source_role.h new file mode 100644 index 0000000..1bfb08d --- /dev/null +++ b/include/sendspin/source_role.h @@ -0,0 +1,137 @@ +// 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 source_role.h +/// @brief Audio capture role that streams audio from the client to the Sendspin server + +#pragma once + +#include "sendspin/config.h" + +#include +#include +#include + +namespace sendspin { + +class SendspinClient; + +// ============================================================================ +// Source types +// ============================================================================ + +/// @brief Line-input signal state reported by the source role in client/state messages +enum class SourceSignal : uint8_t { + PRESENT, // Audio signal detected on the capture input + ABSENT, // No audio signal on the capture input +}; + +/// @brief Listener for source role events. All methods fire on the main loop thread. +class SourceRoleListener { +public: + virtual ~SourceRoleListener() = default; + + /// @brief Called when the outbound stream to the server has opened (the server commanded + /// start and client-stream/start was sent); write_audio() accepts audio from this point on + virtual void on_streaming_started() {} + + /// @brief Called when the outbound stream to the server has closed (server stop command, + /// connection loss, or disconnect); write_audio() rejects audio again + virtual void on_streaming_stopped() {} +}; + +/** + * @brief Audio capture role that streams audio to the server + * + * Streaming is gated by the server (Sendspin spec, Source messages): the client never streams + * unsolicited, the default after connect is stopped, and permission does not survive + * reconnection. When the server commands start, the role announces its capture format with + * client-stream/start and then forwards audio written via write_audio() as timestamped binary + * chunks until the server commands stop (or the connection is lost), closing with + * client-stream/end. + * + * The capture format is fixed at add_source() time: the SourceRoleConfig passed there is the + * format contract for every stream this role opens. Changing formats requires tearing down the + * client and re-adding the role with a new config. + * + * Usage: + * 1. Implement SourceRoleListener to learn when the server starts/stops the stream + * 2. Add the role to the client via SendspinClient::add_source() with the capture format + * 3. Call set_listener() with your listener implementation + * 4. Feed captured audio to write_audio() from your capture thread + * + * @code + * struct MySourceListener : SourceRoleListener { + * void on_streaming_started() override { capture.enable(); } + * void on_streaming_stopped() override { capture.disable(); } + * }; + * + * MySourceListener listener; + * auto& source = client.add_source(SourceRoleConfig{}); + * source.set_listener(&listener); + * + * // Capture thread: + * source.write_audio(pcm_frames, len, capture_time_us); + * @endcode + */ +class SourceRole { + friend class SendspinClient; + +public: + struct Impl; + + SourceRole(SourceRoleConfig config, SendspinClient* client); + ~SourceRole(); + + /// @brief Sets the listener for source events + /// @note The listener must outlive this role + /// @param listener Pointer to the listener implementation + void set_listener(SourceRoleListener* listener); + + /// @brief Writes captured audio into the outbound stream + /// + /// Hot path: non-allocating and never waiting on the consumer (ring timeout 0). The ring is + /// lock-free on ESP; the host implementation takes a short mutex-guarded critical section, + /// so hard-real-time host callers should hand off through their own lock-free stage. + /// Exactly one producer thread may call this; the library does not serialize concurrent + /// writers. + /// + /// @param data Interleaved little-endian signed PCM in the configured format (24-bit as 3 + /// packed bytes per sample). Bytes are forwarded untouched; the config passed to + /// add_source() is the format contract. + /// @param len Length in bytes; must be a whole number of frames (one frame = one sample + /// across all channels). A non-whole-frame write is rejected as a whole. + /// @param capture_time_us Local-clock capture time (same domain as the client's time + /// functions) of the FIRST sample in data. Pass 0 to stamp with the current time, + /// a best-effort fallback for callers that cannot timestamp their ADC. + /// @return 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). + bool write_audio(const uint8_t* data, size_t len, int64_t capture_time_us); + + /// @brief Reports the capture input's signal state, published to the server via client/state + /// + /// Must be called from the main loop thread. Only meaningful when + /// SourceRoleConfig::line_sense is set; ignored (with a warning) otherwise. + /// @param signal The new signal state + void set_signal(SourceSignal signal); + + /// @brief Returns true while the outbound stream is open. Main-loop thread only: reflects + /// the started/stopped listener callbacks, which fire there. + bool is_streaming() const; + +private: + std::unique_ptr impl_; +}; + +} // namespace sendspin diff --git a/src/audio_ring_buffer.cpp b/src/audio_ring_buffer.cpp index 0c3f296..0d4e27f 100644 --- a/src/audio_ring_buffer.cpp +++ b/src/audio_ring_buffer.cpp @@ -25,10 +25,11 @@ static const char* const TAG = "sendspin.ring_buffer"; // Constructor / Destructor // ============================================================================ -std::unique_ptr SendspinAudioRingBuffer::create(size_t buffer_size) { +std::unique_ptr SendspinAudioRingBuffer::create(size_t buffer_size, + MemoryLocation location) { auto rb = std::unique_ptr(new SendspinAudioRingBuffer()); - if (!rb->storage_.allocate(buffer_size)) { + if (!rb->storage_.allocate(buffer_size, location)) { SS_LOGE(TAG, "Failed to allocate %zu bytes for ring buffer", buffer_size); return nullptr; } diff --git a/src/audio_ring_buffer.h b/src/audio_ring_buffer.h index e81d8ae..9fb9e96 100644 --- a/src/audio_ring_buffer.h +++ b/src/audio_ring_buffer.h @@ -50,13 +50,15 @@ struct AudioRingBufferEntry { }; /** - * @brief Pre-allocated SPSC ring buffer for zero-copy encoded audio chunk transfer between - * the network thread and sync task + * @brief Pre-allocated SPSC ring buffer for zero-copy timestamped audio chunk transfer + * between one producer thread and one consumer thread * * Each entry stores an AudioRingBufferEntry header followed immediately by the variable- * length audio payload. All storage is pre-allocated at construction; there are no per- - * chunk heap allocations. The SPSC contract is: one producer (WebSocket callback thread) - * calls write_chunk(), and one consumer (sync task) calls receive_chunk() / return_chunk(). + * chunk heap allocations. The SPSC contract is: exactly one producer calls write_chunk(), + * and exactly one consumer calls receive_chunk() / return_chunk(). The player uses it + * inbound (network thread -> sync task, encoded audio); the source uses it outbound (the + * consumer's capture thread -> source task, captured PCM). * * Usage: * 1. Call create() to allocate and initialize the ring buffer @@ -83,15 +85,18 @@ class SendspinAudioRingBuffer { public: /// @brief Creates a ring buffer with the specified total storage size. /// @param buffer_size Total ring buffer storage in bytes. + /// @param location Memory placement for the storage (ESP-IDF only; ignored on host). /// @return unique_ptr to the ring buffer, or nullptr on allocation failure. - static std::unique_ptr create(size_t buffer_size); + static std::unique_ptr create( + size_t buffer_size, MemoryLocation location = MemoryLocation::PREFER_EXTERNAL); ~SendspinAudioRingBuffer(); /// @brief Writes an audio chunk into the ring buffer. /// @param data Pointer to the audio data. /// @param data_size Size of the audio data in bytes. - /// @param timestamp Server timestamp for this chunk. + /// @param timestamp Producer-defined timestamp for this chunk; the time domain is the + /// producer's contract (server play time inbound, local capture time outbound). /// @param chunk_type Type of audio chunk. /// @param timeout_ms Milliseconds to wait if buffer is full (UINT32_MAX = wait forever). /// @return true if successfully written, false if buffer full or error. diff --git a/src/client.cpp b/src/client.cpp index 0f2c27c..b5be1f7 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -37,6 +37,9 @@ #ifdef SENDSPIN_ENABLE_PLAYER #include "player_role_impl.h" #endif +#ifdef SENDSPIN_ENABLE_SOURCE +#include "source_role_impl.h" +#endif #include "protocol_messages.h" #ifdef SENDSPIN_ENABLE_VISUALIZER #include "visualizer_role_impl.h" @@ -82,10 +85,14 @@ 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 // event_state_'s Inbox on destruction, so all roles must be gone before the alphabetized - // member order destroys event_state_. + // member order destroys event_state_. The source role additionally holds a raw + // ConnectionManager* for its task, so it must be gone before connection_manager_ below. #ifdef SENDSPIN_ENABLE_PLAYER this->player_.reset(); #endif +#ifdef SENDSPIN_ENABLE_SOURCE + this->source_.reset(); +#endif #ifdef SENDSPIN_ENABLE_VISUALIZER this->visualizer_.reset(); #endif @@ -146,6 +153,14 @@ bool SendspinClient::start_server() { } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_) { + if (!this->source_->impl_->start()) { + return false; + } + } +#endif + // Create and configure the WebSocket server (started later when network is ready) this->connection_manager_->init_server(this); @@ -302,6 +317,16 @@ void SendspinClient::loop() { this->visualizer_->impl_->handle_stream_ring_event( static_cast(event.code)); } +#endif + break; + } + // Appended to the role's pending events; dispatched by this tick's drain + case InboxEventType::SOURCE_STREAM: { +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_) { + this->source_->impl_->on_stream_ring_event( + static_cast(event.code)); + } #endif break; } @@ -357,6 +382,11 @@ void SendspinClient::loop() { this->artwork_->impl_->drain_events(); } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_ && this->source_->impl_->needs_drain(slot_bits)) { + this->source_->impl_->drain_events(); + } +#endif // --- Group update events --- if (slot_bits & INBOX_TOPIC_GROUP) { @@ -461,6 +491,19 @@ VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { } #endif +#ifdef SENDSPIN_ENABLE_SOURCE +SourceRole& SendspinClient::add_source(SourceRoleConfig config) { + if (this->started_) { + SS_LOGW(TAG, "add_source() called after start_server()"); + } + this->source_ = std::make_unique(config, this); + this->source_->impl_->attach_inbox(this->event_state_->inbox); + // Only SendspinClient can reach the manager; it outlives the role (destructor order) + this->source_->impl_->connection_manager = this->connection_manager_.get(); + return *this->source_; +} +#endif + // ============================================================================ // Queries // ============================================================================ @@ -590,6 +633,11 @@ void SendspinClient::cleanup_connection_state() { this->visualizer_->impl_->cleanup(); } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_) { + this->source_->impl_->cleanup(); + } +#endif // Release high-performance networking for time sync if (this->high_performance_held_for_time_) { @@ -655,6 +703,11 @@ std::string SendspinClient::build_hello_message() { this->visualizer_->impl_->build_hello_fields(msg); } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_) { + this->source_->impl_->build_hello_fields(msg); + } +#endif return format_client_hello_message(&msg); } @@ -880,6 +933,22 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* } } #endif + +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_ != nullptr && conn != nullptr && + this->connection_manager_->current_shared().get() == conn) { + // Only the current connection may write the command slot: a nursery or + // handoff-displaced connection delivering concurrently must not overwrite the + // current connection's latest command (per-connection permission; Sendspin + // spec, Source messages). The instance id lets the main-loop latch discard the + // remaining TOCTOU sliver where the writer is displaced before the drain. + SourceCommand source_cmd{}; + if (process_server_command_source(root, &source_cmd)) { + this->source_->impl_->handle_server_command(source_cmd, + conn->get_instance_id()); + } + } +#endif break; } case SendspinServerToClientMessageType::GROUP_UPDATE: { @@ -970,6 +1039,12 @@ void SendspinClient::publish_client_state(SendspinConnection* conn) { } #endif +#ifdef SENDSPIN_ENABLE_SOURCE + if (this->source_) { + this->source_->impl_->build_state_fields(state_msg); + } +#endif + std::string state_message = format_client_state_message(&state_msg); conn->send_text_message(state_message, nullptr); } diff --git a/src/constants.h b/src/constants.h index 5655015..7a0aee9 100644 --- a/src/constants.h +++ b/src/constants.h @@ -13,7 +13,7 @@ // limitations under the License. /// @file constants.h -/// @brief Shared constants for unit conversions +/// @brief Shared constants for unit conversions and cross-task timing #pragma once @@ -25,4 +25,9 @@ static constexpr int64_t US_PER_MS = 1000LL; static constexpr uint32_t MS_PER_SECOND = 1000U; static constexpr uint32_t US_PER_SECOND = 1000000U; +/// @brief Wait time (ms) between retries while a task waits for the time filter's first +/// measurement (SendspinClient::is_time_synced()). Shared by the sync and source tasks so both +/// audio pipelines gate on time sync with the same cadence. +static constexpr uint32_t WAIT_FOR_TIME_SYNC_MS = 15U; + } // namespace sendspin diff --git a/src/inbox.h b/src/inbox.h index 2f031f0..3525574 100644 --- a/src/inbox.h +++ b/src/inbox.h @@ -48,6 +48,7 @@ static constexpr uint32_t INBOX_TOPIC_PLAYER_STREAM_PARAMS = 1U << 6; // Player static constexpr uint32_t INBOX_TOPIC_VISUALIZER_CONFIG = 1U << 7; // Visualizer config slot static constexpr uint32_t INBOX_TOPIC_ARTWORK_DISPLAY = 1U << 8; // Artwork display slot static constexpr uint32_t INBOX_TOPIC_PLAYER_STATE = 1U << 9; // Player client-state slot +static constexpr uint32_t INBOX_TOPIC_SOURCE_COMMAND = 1U << 10; // Source command slot // ============================================================================ // Event ring types @@ -65,6 +66,7 @@ enum class InboxEventType : uint8_t { COLOR_CLEARED, // Color state cleared on disconnect; no payload ARTWORK_STREAM, // Artwork stream lifecycle; code = ArtworkEventType VISUALIZER_STREAM, // Visualizer stream lifecycle; code = VisualizerEventType + SOURCE_STREAM, // Source stream lifecycle; code = SourceStreamCallbackType }; /// @brief Payload for TIME_RESPONSE events diff --git a/src/protocol_messages.h b/src/protocol_messages.h index 18745ba..844ca09 100644 --- a/src/protocol_messages.h +++ b/src/protocol_messages.h @@ -23,6 +23,7 @@ #include "sendspin/controller_role.h" #include "sendspin/metadata_role.h" #include "sendspin/player_role.h" +#include "sendspin/source_role.h" #include "sendspin/types.h" #include "sendspin/visualizer_role.h" #include @@ -654,12 +655,6 @@ inline std::optional source_command_from_string(const std::string return std::nullopt; } -/// @brief Line-input signal state reported by the source role in client/state messages -enum class SourceSignal : uint8_t { - PRESENT, // Audio signal detected on the capture input - ABSENT, // No audio signal on the capture input -}; - inline const char* to_cstr(SourceSignal signal) { switch (signal) { case SourceSignal::PRESENT: diff --git a/src/source_encoder.h b/src/source_encoder.h new file mode 100644 index 0000000..3c17caf --- /dev/null +++ b/src/source_encoder.h @@ -0,0 +1,78 @@ +// 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 source_encoder.h +/// @brief Encoder seam between the source task's PCM chunk assembly and the wire payload + +#pragma once + +#include +#include +#include + +namespace sendspin { + +/// @brief Turns one assembled chunk of capture PCM into the outbound wire payload, keeping the +/// source task's chunk loop codec-agnostic +class SourceEncoder { +public: + virtual ~SourceEncoder() = default; + + /// @brief Whether a chunk of `in_len` PCM bytes is encodable; only gates the stream-end + /// remainder (mid-stream chunks are full by config validation), which the task skips rather + /// than pads when this returns false -- a final short chunk is optional per the spec + virtual bool can_encode(size_t in_len) const { + return in_len > 0; + } + + /// @brief Encodes one chunk of interleaved PCM into the wire payload area + /// + /// @param in Assembled PCM chunk. `in == out` (exact aliasing) is part of the contract: + /// the task assembles directly into the send buffer and encodes in place. + /// @param in_len Length of the PCM chunk in bytes. + /// @param out Destination payload area. + /// @param out_capacity Capacity of `out` in bytes. + /// @return Number of payload bytes written to `out`; 0 on encode failure (the task drops + /// the chunk). + virtual size_t encode(const uint8_t* in, size_t in_len, uint8_t* out, size_t out_capacity) = 0; + + /// @brief Algorithmic delay (µs), subtracted from each chunk's capture anchor so the wire + /// timestamp names the audio the payload actually carries + virtual int64_t lookahead_us() const = 0; + + /// @brief Resets encoder state between streams + virtual void reset() = 0; +}; + +/// @brief Identity encoder for PCM streams: the assembled chunk already is the wire payload +class PcmPassthroughEncoder final : public SourceEncoder { +public: + size_t encode(const uint8_t* in, size_t in_len, uint8_t* out, size_t out_capacity) override { + if (in_len > out_capacity) { + return 0; // Only reachable in the separate-buffer shape of the contract + } + if (in != out) { + memcpy(out, in, in_len); + } + return in_len; + } + + int64_t lookahead_us() const override { + return 0; + } + + void reset() override {} +}; + +} // namespace sendspin diff --git a/src/source_role.cpp b/src/source_role.cpp new file mode 100644 index 0000000..4b78b11 --- /dev/null +++ b/src/source_role.cpp @@ -0,0 +1,286 @@ +// 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. + +#include "connection.h" +#include "connection_manager.h" +#include "platform/logging.h" +#include "protocol_messages.h" +#include "sendspin/client.h" +#include "source_role_impl.h" + +static const char* const TAG = "sendspin.source"; + +namespace sendspin { + +// ============================================================================ +// Helpers +// ============================================================================ + +/// @brief Validates the capture format contract, logging every rejection at ERROR. Fail +/// closed, never clamp or repair; an invalid config leaves the role inert (added, but never +/// advertised or streaming), like a player with no audio formats. +static bool validate_config(const SourceRoleConfig& config) { + bool valid = true; + if (config.codec != SendspinCodecFormat::PCM) { + // OPUS is a recognized config value but not implemented yet; anything else is invalid + // outright. Both reject the same way (fail closed). + SS_LOGE(TAG, "Rejecting source config: only the pcm codec is supported (got %d)", + static_cast(config.codec)); + valid = false; + } + if (config.sample_rate == 0) { + SS_LOGE(TAG, "Rejecting source config: sample_rate must be > 0"); + valid = false; + } + if (config.channels == 0) { + SS_LOGE(TAG, "Rejecting source config: channels must be > 0"); + valid = false; + } + if (config.bit_depth != 16 && config.bit_depth != 24 && config.bit_depth != 32) { + SS_LOGE(TAG, "Rejecting source config: bit_depth must be 16, 24, or 32 (got %u)", + config.bit_depth); + valid = false; + } + if (config.chunk_duration_ms < SourceRoleConfig::CHUNK_MIN_MS || + config.chunk_duration_ms > SourceRoleConfig::CHUNK_MAX_MS) { + // Sendspin spec, Source messages: MUST <= 150 ms, SHOULD >= 5 ms; the SHOULD is + // deliberately enforced as hard as the MUST + SS_LOGE(TAG, "Rejecting source config: chunk_duration_ms %u outside [%u, %u]", + config.chunk_duration_ms, SourceRoleConfig::CHUNK_MIN_MS, + SourceRoleConfig::CHUNK_MAX_MS); + valid = false; + } + if (config.capture_buffer_ms == 0) { + SS_LOGE(TAG, "Rejecting source config: capture_buffer_ms must be > 0"); + valid = false; + } + return valid; +} + +// ============================================================================ +// Impl constructor / destructor +// ============================================================================ + +SourceRole::Impl::Impl(SourceRoleConfig config, SendspinClient* client) + : config(config), + client(client), + event_state(std::make_unique()), + task(std::make_unique()), + config_valid(validate_config(config)) {} + +SourceRole::Impl::~Impl() { + // Join the task first: the consumer's capture thread may still be inside write_audio(), + // so the task and its capture ring must outlive that call + this->task.reset(); +} + +// ============================================================================ +// SourceRole forwarding (public API -> Impl) +// ============================================================================ + +SourceRole::SourceRole(SourceRoleConfig config, SendspinClient* client) + : impl_(std::make_unique(config, client)) {} + +SourceRole::~SourceRole() = default; + +void SourceRole::set_listener(SourceRoleListener* listener) { + this->impl_->listener = listener; +} + +bool SourceRole::write_audio(const uint8_t* data, size_t len, int64_t capture_time_us) { + return this->impl_->write_audio(data, len, capture_time_us); +} + +void SourceRole::set_signal(SourceSignal signal) { + this->impl_->set_signal(signal); +} + +bool SourceRole::is_streaming() const { + return this->impl_->streaming_active; +} + +// ============================================================================ +// Impl: Internal integration methods +// ============================================================================ + +void SourceRole::Impl::attach_inbox(Inbox& inbox) { + this->inbox = &inbox; + this->event_state->command_slot.bind(inbox, INBOX_TOPIC_SOURCE_COMMAND); +} + +bool SourceRole::Impl::start() { + if (this->config_valid && !this->task->is_initialized()) { + if (!this->task->init(this, this->client, this->connection_manager)) { + SS_LOGE(TAG, "Failed to initialize source task"); + return false; + } + if (!this->task->start(this->config.psram_stack, this->config.priority)) { + SS_LOGE(TAG, "Failed to start source task thread"); + return false; + } + } + return true; +} + +void SourceRole::Impl::build_hello_fields(ClientHelloMessage& msg) { + if (!this->config_valid) { + return; + } + + msg.supported_roles.push_back(SendspinRole::SOURCE); + msg.source_v1_support = SourceSupportObject{.line_sense = this->config.line_sense}; +} + +void SourceRole::Impl::build_state_fields(ClientStateMessage& msg) const { + if (!this->config_valid) { + return; + } + + // The source object may legitimately be present and empty; signal is included only when + // line_sense is configured and a state has been reported (Sendspin spec, Source messages -- + // Client state object) + ClientSourceStateObject source_state{}; + if (this->config.line_sense && this->signal.has_value()) { + source_state.signal = this->signal; + } + msg.source = source_state; +} + +void SourceRole::Impl::handle_server_command(SourceCommand command, + uint64_t connection_instance_id) const { + this->event_state->command_slot.write(SourceCommandEnvelope{connection_instance_id, command}); +} + +void SourceRole::Impl::on_stream_ring_event(SourceStreamCallbackType event) { + this->pending_events.push_back(event); +} + +void SourceRole::Impl::drain_events() { + // --- Server command latch --- + SourceCommandEnvelope envelope; + if (this->event_state->command_slot.take(envelope)) { + auto* current = this->connection_manager->current(); + if (current == nullptr || current->get_instance_id() != envelope.connection_instance_id) { + // Per-connection permission (Sendspin spec, Source messages): a command from a + // displaced connection must not grant the current one + SS_LOGD(TAG, "Discarding source command from a stale connection"); + } else { + const bool start = envelope.command == SourceCommand::START; + // Only a transition reaches the task: commands are idempotent per the spec + if (start != this->streaming_desired) { + this->streaming_desired = start; + if (start) { + this->task->signal_start(); + } else { + this->task->signal_stop(); + } + } + } + } + + // --- Stream lifecycle events from the task --- + if (this->pending_events.empty()) { + return; + } + + size_t processed = 0; + // Indexed with a fresh size() check per iteration: a listener callback may re-enter + // cleanup(), which clears this vector mid-loop (player-role pattern) + // NOLINTNEXTLINE(modernize-loop-convert): body mutates the vector, see above + for (size_t idx = 0; idx < this->pending_events.size(); ++idx) { + const SourceStreamCallbackType event = this->pending_events[idx]; + switch (event) { + case SourceStreamCallbackType::STREAMING_STARTED: { + if (!this->streaming_active) { + this->streaming_active = true; + if (this->listener) { + this->listener->on_streaming_started(); + } + } + break; + } + case SourceStreamCallbackType::STREAMING_STOPPED: { + // streaming_active keeps the callbacks paired 1:1 (cleanup() enqueues an + // unconditional STOPPED) + if (this->streaming_active) { + this->streaming_active = false; + if (this->listener) { + this->listener->on_streaming_stopped(); + } + } + break; + } + } + ++processed; + } + + // A re-entrant cleanup() may have cleared the vector mid-loop + if (processed > this->pending_events.size()) { + processed = this->pending_events.size(); + } + if (processed > 0) { + this->pending_events.erase( + this->pending_events.begin(), + this->pending_events.begin() + static_cast(processed)); + } +} + +void SourceRole::Impl::cleanup() { + // Permission does not survive the connection (Sendspin spec, Source messages). The gate is + // closed here, before the synthetic STOPPED below, so the listener can never observe + // "stopped" while write_audio() still accepts. + this->streaming_desired = false; + this->task->close_audio_gate(); + this->task->signal_stop(); + + // Discard a stale command from the dead connection + this->event_state->command_slot.reset(); + + // STREAMING_STOPPED is pushed unconditionally: a second teardown in the same tick wipes the + // ring before it drains (see cleanup_connection_state()); streaming_active keeps the + // listener callbacks paired, so an extra STOPPED is harmless + this->pending_events.clear(); + this->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STOPPED); +} + +// ============================================================================ +// Impl: Consumer-facing method implementations +// ============================================================================ + +bool SourceRole::Impl::write_audio(const uint8_t* data, size_t len, int64_t capture_time_us) const { + return this->task->write_audio(data, len, capture_time_us); +} + +void SourceRole::Impl::set_signal(SourceSignal new_signal) { + if (!this->config.line_sense) { + SS_LOGW(TAG, "set_signal() ignored: line_sense not configured"); + return; + } + this->signal = new_signal; + this->client->publish_state(); +} + +// ============================================================================ +// Impl: Helpers +// ============================================================================ + +void SourceRole::Impl::enqueue_stream_event(SourceStreamCallbackType event) const { + // A dropped lifecycle event wedges the listener state, so drops log at ERROR + push_event_or_log(this->inbox, InboxEventType::SOURCE_STREAM, static_cast(event), TAG, + event == SourceStreamCallbackType::STREAMING_STARTED ? "STREAMING_STARTED" + : "STREAMING_STOPPED", + /*error_level=*/true); +} + +} // namespace sendspin diff --git a/src/source_role_impl.h b/src/source_role_impl.h new file mode 100644 index 0000000..38e58c1 --- /dev/null +++ b/src/source_role_impl.h @@ -0,0 +1,121 @@ +// 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 source_role_impl.h +/// @brief Private implementation for the source role (pimpl) + +#pragma once + +#include "inbox.h" +#include "protocol_messages.h" +#include "sendspin/source_role.h" +#include "source_task.h" + +#include +#include +#include + +namespace sendspin { + +class ConnectionManager; +class SendspinClient; + +/// @brief Deferred stream lifecycle callback types queued from the source task thread +enum class SourceStreamCallbackType : uint8_t { + STREAMING_STARTED, // The outbound stream opened (client-stream/start sent) + STREAMING_STOPPED, // The outbound stream closed (stop, connection loss, or cleanup) +}; + +/// @brief Private implementation of the source role +struct SourceRole::Impl { + Impl(SourceRoleConfig config, SendspinClient* client); + ~Impl(); + + // ======================================== + // Event state + // ======================================== + + /// @brief One server/command source command with the connection it arrived on + struct SourceCommandEnvelope { + uint64_t connection_instance_id{0}; + SourceCommand command{SourceCommand::STOP}; + }; + + struct EventState { + /// Latest-wins: commands are idempotent and only the final desired state matters + /// (Sendspin spec, Source messages) + InboxSlot command_slot; + }; + + // ======================================== + // Internal integration methods (called by SendspinClient) + // ======================================== + + void attach_inbox(Inbox& inbox); + bool start(); + void build_hello_fields(ClientHelloMessage& msg); + void build_state_fields(ClientStateMessage& msg) const; + void handle_server_command(SourceCommand command, uint64_t connection_instance_id) const; + void on_stream_ring_event(SourceStreamCallbackType event); + // A pending server command, or lifecycle events appended during this tick's ring dispatch + bool needs_drain(uint32_t pending_bits) const { + return (pending_bits & INBOX_TOPIC_SOURCE_COMMAND) != 0 || !this->pending_events.empty(); + } + void drain_events(); + void cleanup(); + + // ======================================== + // Consumer-facing method implementations + // ======================================== + + bool write_audio(const uint8_t* data, size_t len, int64_t capture_time_us) const; + void set_signal(SourceSignal new_signal); + + // ======================================== + // Helpers + // ======================================== + + void enqueue_stream_event(SourceStreamCallbackType event) const; + + // ======================================== + // Fields + // ======================================== + + // Struct fields + SourceRoleConfig config; + std::vector pending_events; + /// Last set_signal() value, included in client/state once set (main-thread only) + std::optional signal; + + // Pointer fields + SendspinClient* client; + /// Set by add_source(); used for the instance-id check and stream binding. Outlives this + /// role: SendspinClient destroys every role before connection_manager_. + ConnectionManager* connection_manager{nullptr}; + std::unique_ptr event_state; + Inbox* inbox{nullptr}; + SourceRoleListener* listener{nullptr}; + std::unique_ptr task; + + // 8-bit fields + /// Set once in the constructor; an invalid config leaves the role inert + bool config_valid{false}; + /// Main-loop latch of the server's last command on the current connection; cleanup() + /// resets it to stopped (per-connection permission, Sendspin spec, Source messages) + bool streaming_desired{false}; + /// Keeps the listener's started/stopped callbacks paired 1:1 (main-thread only) + bool streaming_active{false}; +}; + +} // namespace sendspin diff --git a/src/source_task.cpp b/src/source_task.cpp new file mode 100644 index 0000000..a2fa39b --- /dev/null +++ b/src/source_task.cpp @@ -0,0 +1,465 @@ +// 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. + +#include "source_task.h" + +#include "audio_types.h" +#include "connection.h" +#include "connection_manager.h" +#include "platform/logging.h" +#include "platform/thread.h" +#include "platform/time.h" +#include "protocol_messages.h" +#include "source_role_impl.h" +#include "time_filter.h" + +#include +#include +#include + +namespace sendspin { + +static const char* const TAG = "sendspin.source_task"; + +/// @brief Same budget as the sync task. Host -O2 -fstack-usage measures the deepest task-path +/// chain (stream -> send_chunk -> event wait) near 0.6 KB; the budget's remainder is headroom +/// for the ESP transport send path and platform differences, pending an on-target high-water +/// measurement +static constexpr size_t SOURCE_TASK_STACK_SIZE = 6192; + +/// @brief Ring receive timeout (ms) bounding how long the task waits before re-checking the +/// stop/connection conditions; same cadence as the sync task's encoded-chunk receive +static constexpr uint32_t CAPTURE_RECEIVE_TIMEOUT_MS = 15U; + +/// @brief Binary type byte plus the BE64 capture timestamp (Sendspin spec, Source messages) +static constexpr size_t SOURCE_WIRE_HEADER_SIZE = 1U + BINARY_TIMESTAMP_SIZE; + +/// @brief Backoff (ms) before retrying a failed stream open on a still-live connection; coarse +/// because opens are lifecycle-rare and the failure means transient pressure needing time +static constexpr uint32_t SOURCE_OPEN_RETRY_MS = 500U; + +/// @brief Bound (ms) on waiting for the client-stream/start send confirmation. Generous next to +/// any healthy queue drain, and safe because the text completion is best-effort: a skipped +/// callback times out here and the open is retried rather than the task wedging +static constexpr uint32_t START_CONFIRM_TIMEOUT_MS = 2000U; + +/// @brief Ring metadata margin: +1/4 over audio capacity for per-write entry headers, the +/// inverse of the player's 1/5 advertise fraction (AUDIO_BUFFER_ADVERTISE_DENOMINATOR) +static constexpr size_t CAPTURE_RING_OVERHEAD_DENOMINATOR = 4U; + +// ============================================================================ +// Lifecycle +// ============================================================================ + +SourceTask::~SourceTask() { + this->stop(); +} + +bool SourceTask::init(SourceRole::Impl* source_impl, SendspinClient* client, + ConnectionManager* connections) { + this->source_impl_ = source_impl; + this->client_ = client; + this->connections_ = connections; + + const SourceRoleConfig& config = source_impl->config; + this->bytes_per_frame_ = source_bytes_per_frame(config.channels, config.bit_depth); + + // 64-bit intermediates with a fail-closed cap: ms x rate x bytes-per-frame can wrap a + // 32-bit size_t, and a wrapped size would be a repaired config rather than a rejected one + constexpr uint64_t MAX_BUFFER_BYTES = UINT32_MAX / 2; + const uint64_t chunk_bytes = + source_ms_to_frames(config.chunk_duration_ms, config.sample_rate) * this->bytes_per_frame_; + if (chunk_bytes == 0 || chunk_bytes > MAX_BUFFER_BYTES) { + // Zero: a sample rate so low the chunk duration holds no whole frame + SS_LOGE(TAG, "Source chunk of %u ms at %u Hz is unusable (%llu bytes)", + config.chunk_duration_ms, config.sample_rate, + static_cast(chunk_bytes)); + return false; + } + this->chunk_bytes_ = static_cast(chunk_bytes); + + const uint64_t audio_bytes = + source_ms_to_frames(config.capture_buffer_ms, config.sample_rate) * this->bytes_per_frame_; + if (audio_bytes > MAX_BUFFER_BYTES) { + SS_LOGE(TAG, "Source capture buffer of %u ms at %u Hz is too large (%llu bytes)", + config.capture_buffer_ms, config.sample_rate, + static_cast(audio_bytes)); + return false; + } + + if (!this->event_flags_.create()) { + SS_LOGE(TAG, "Couldn't create event flags."); + return false; + } + + this->capture_ring_ = SendspinAudioRingBuffer::create( + static_cast(audio_bytes + audio_bytes / CAPTURE_RING_OVERHEAD_DENOMINATOR), + config.buffer_location); + if (this->capture_ring_ == nullptr) { + SS_LOGE(TAG, "Couldn't create capture ring buffer."); + return false; + } + + if (!this->staging_.allocate(SOURCE_WIRE_HEADER_SIZE + this->chunk_bytes_, + config.buffer_location)) { + SS_LOGE(TAG, "Couldn't allocate chunk staging buffer."); + return false; + } + + // Config validation admits only PCM, so the passthrough is the only encoder to construct; + // the seam exists for the Opus encoder to slot in (see source_encoder.h) + this->encoder_ = std::make_unique(); + + this->send_complete_cb_ = [this](bool ok) { + this->last_send_ok_.store(ok, std::memory_order_release); + this->event_flags_.set(SourceTaskBits::SOURCE_SEND_COMPLETE); + }; + + return true; +} + +bool SourceTask::start(bool task_stack_in_psram, unsigned priority) { + if (!this->is_initialized()) { + SS_LOGE(TAG, "Source task not initialized (call init() first)"); + return false; + } + + if (this->task_thread_.joinable()) { + SS_LOGW(TAG, "Source task thread already started"); + return false; + } + + this->event_flags_.clear( + SourceTaskBits::SOURCE_TASK_RUNNING | SourceTaskBits::SOURCE_TASK_STOPPED | + SourceTaskBits::SOURCE_TASK_IDLE | SourceTaskBits::SOURCE_COMMAND_STOP | + SourceTaskBits::SOURCE_COMMAND_UPDATE | SourceTaskBits::SOURCE_SEND_COMPLETE); + + platform_configure_thread("SsSrc", SOURCE_TASK_STACK_SIZE, static_cast(priority), + task_stack_in_psram); + + this->task_thread_ = std::thread(thread_entry, this); + + // Wait for the thread to reach IDLE before returning + this->event_flags_.wait(SourceTaskBits::SOURCE_TASK_IDLE | SourceTaskBits::SOURCE_TASK_STOPPED, + false, false, UINT32_MAX); + + return true; +} + +// ============================================================================ +// Public API +// ============================================================================ + +void SourceTask::signal_start() { + // Setting bits on uncreated event flags is a null-handle crash on ESP; init() only runs + // for a valid config, so every signal/query path guards on is_initialized() + if (!this->is_initialized()) { + return; + } + this->stream_requested_.store(true, std::memory_order_release); + this->event_flags_.set(SourceTaskBits::SOURCE_COMMAND_UPDATE); +} + +void SourceTask::signal_stop() { + if (!this->is_initialized()) { + return; + } + this->stream_requested_.store(false, std::memory_order_release); + this->event_flags_.set(SourceTaskBits::SOURCE_COMMAND_UPDATE); +} + +bool SourceTask::write_audio(const uint8_t* data, size_t len, int64_t capture_time_us) { + // Defaults false and is only set by a running task, so a never-initialized role rejects too + if (!this->accepting_audio_.load(std::memory_order_acquire)) { + return false; + } + if (data == nullptr || len == 0 || (len % this->bytes_per_frame_) != 0U) { + // A forwarded partial frame would shift the channel interleaving of every later sample; + // warn once per episode so a misconfigured caller cannot flood the log at capture rate + if (!this->producer_frame_warned_) { + this->producer_frame_warned_ = true; + SS_LOGW(TAG, "write_audio() rejected: %u bytes is not a whole number of %u-byte frames", + static_cast(len), static_cast(this->bytes_per_frame_)); + } + return false; + } + if (capture_time_us == 0) { + // Best-effort arrival stamp for callers that cannot timestamp their ADC + capture_time_us = platform_time_us(); + } + if (!this->capture_ring_->write_chunk(data, len, capture_time_us, CHUNK_TYPE_ENCODED_AUDIO, + 0)) { + // The ring bounds the stall backlog (capture_buffer_ms); warn once per overflow + // episode, the recovery log carries the total + ++this->producer_dropped_writes_; + if (!this->producer_drop_episode_) { + this->producer_drop_episode_ = true; + SS_LOGW(TAG, "Capture ring full; dropping writes until it drains"); + } + return false; + } + if (this->producer_drop_episode_) { + this->producer_drop_episode_ = false; + SS_LOGI(TAG, "Capture resumed after dropping %u writes", this->producer_dropped_writes_); + this->producer_dropped_writes_ = 0; + } + this->producer_frame_warned_ = false; + return true; +} + +// ============================================================================ +// Task loop +// ============================================================================ + +void SourceTask::thread_entry(void* params) { + auto* task = static_cast(params); + task->run(); +} + +void SourceTask::run() { + this->event_flags_.set(SourceTaskBits::SOURCE_TASK_IDLE); + + while (true) { + const uint32_t flags = this->event_flags_.wait( + SourceTaskBits::SOURCE_COMMAND_STOP | SourceTaskBits::SOURCE_COMMAND_UPDATE, false, + false, UINT32_MAX); + if ((flags & SourceTaskBits::SOURCE_COMMAND_STOP) != 0U) { + break; + } + this->event_flags_.clear(SourceTaskBits::SOURCE_COMMAND_UPDATE); + if (!this->stream_requested_.load(std::memory_order_acquire)) { + continue; // A stop that raced an earlier start; nothing to do while idle + } + + // Streaming permission is per-connection (Sendspin spec, Source messages): bind the + // stream to one connection for its whole life; a swap ends it rather than migrating it + auto conn = this->connections_->current_shared(); + if (conn == nullptr) { + continue; // Vanished since the latch; the role's cleanup wakes this loop again + } + + this->event_flags_.clear(SourceTaskBits::SOURCE_TASK_IDLE); + this->event_flags_.set(SourceTaskBits::SOURCE_TASK_RUNNING); + this->stream(conn); + this->event_flags_.clear(SourceTaskBits::SOURCE_TASK_RUNNING); + this->event_flags_.set(SourceTaskBits::SOURCE_TASK_IDLE); + + // stream() exiting with desired-state still streaming on a still-current connection + // means the open itself failed (every other exit clears one of these conditions); + // the server will not repeat its start, so back off and re-attempt instead of parking + if ((this->event_flags_.get() & SourceTaskBits::SOURCE_COMMAND_STOP) == 0U && + this->stream_requested_.load(std::memory_order_acquire) && + this->connections_->current_shared() == conn) { + this->event_flags_.wait( + SourceTaskBits::SOURCE_COMMAND_STOP | SourceTaskBits::SOURCE_COMMAND_UPDATE, false, + false, SOURCE_OPEN_RETRY_MS); + this->event_flags_.set(SourceTaskBits::SOURCE_COMMAND_UPDATE); + } + } + + this->event_flags_.clear(SourceTaskBits::SOURCE_TASK_IDLE | + SourceTaskBits::SOURCE_TASK_RUNNING); + this->event_flags_.set(SourceTaskBits::SOURCE_TASK_STOPPED); +} + +void SourceTask::stream(const std::shared_ptr& conn) { + if (!this->wait_for_time_sync(conn)) { + return; // Cancelled while waiting; nothing was sent, so nothing to close + } + + // No negotiation: the add_source() config is the announced format (Sendspin spec, Source + // messages -- client-stream/start). JSON on this thread is fine, open/close is rare. + const SourceRoleConfig& config = this->source_impl_->config; + ClientStreamStartMessage start_msg; + start_msg.codec = config.codec; + start_msg.channels = config.channels; + start_msg.sample_rate = config.sample_rate; + start_msg.bit_depth = config.bit_depth; + // A clean slate for the confirmation wait: a completion left over from a prior stream's + // timed-out send on this connection could otherwise satisfy the wait below spuriously. + this->event_flags_.clear(SourceTaskBits::SOURCE_SEND_COMPLETE); + if (conn->send_text_message(format_client_stream_start_message(&start_msg), + this->send_complete_cb_) != SsErr::OK) { + SS_LOGW(TAG, "Failed to send client-stream/start; stream not opened"); + return; + } + // Queued is not sent: on the async ESP-server transport OK only means httpd accepted the + // work, and the worker can still fail the wire send. Chunks MUST follow a delivered + // client-stream/start (Sendspin spec, Source messages), so wait for the completion before + // opening the gate. The text callback is best-effort (it can be skipped on teardown), hence + // the bounded wait; a timeout or failure is treated as a failed open and retried. + const uint32_t bits = this->event_flags_.wait(SourceTaskBits::SOURCE_SEND_COMPLETE, false, true, + START_CONFIRM_TIMEOUT_MS); + if ((bits & SourceTaskBits::SOURCE_SEND_COMPLETE) == 0U || + !this->last_send_ok_.load(std::memory_order_acquire)) { + SS_LOGW(TAG, "client-stream/start not confirmed; stream not opened"); + return; + } + + this->encoder_->reset(); + this->chunk_filled_bytes_ = 0; + this->stall_episode_ = false; + this->stall_flushed_entries_ = 0; + + // Flush while the gate is still closed (no writer can race it), so the first chunk is live + // audio by construction; then open for capture -- chunks must follow client-stream/start + this->flush_ring_to_live(); + this->accepting_audio_.store(true, std::memory_order_release); + this->source_impl_->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STARTED); + + while (this->stream_still_open(conn)) { + if (this->current_entry_ == nullptr) { + this->current_entry_ = this->capture_ring_->receive_chunk(CAPTURE_RECEIVE_TIMEOUT_MS); + this->entry_consumed_bytes_ = 0; + if (this->current_entry_ == nullptr) { + continue; // No capture yet; re-check the stream conditions + } + } + + if (this->chunk_filled_bytes_ == 0) { + // Anchor on the chunk's own first sample, past any already-consumed entry frames + this->chunk_anchor_us_ = + source_entry_anchor_us(this->current_entry_->timestamp, this->entry_consumed_bytes_, + this->bytes_per_frame_, config.sample_rate); + } + + uint8_t* payload = this->staging_.data() + SOURCE_WIRE_HEADER_SIZE; + const size_t take = std::min(this->current_entry_->data_size - this->entry_consumed_bytes_, + this->chunk_bytes_ - this->chunk_filled_bytes_); + memcpy(payload + this->chunk_filled_bytes_, + this->current_entry_->data() + this->entry_consumed_bytes_, take); + this->chunk_filled_bytes_ += take; + this->entry_consumed_bytes_ += take; + if (this->entry_consumed_bytes_ == this->current_entry_->data_size) { + this->capture_ring_->return_chunk(this->current_entry_); + this->current_entry_ = nullptr; + } + + if (this->chunk_filled_bytes_ == this->chunk_bytes_) { + const bool sent = this->send_chunk(conn); + this->chunk_filled_bytes_ = 0; + if (!sent) { + // Drop the backlog and resume from live capture rather than bursting stale + // audio (Sendspin spec, Source messages -- stall policy); warn once per episode + this->stall_flushed_entries_ += this->flush_ring_to_live(); + if (!this->stall_episode_) { + this->stall_episode_ = true; + SS_LOGW(TAG, "Source send stalled; dropping to live capture"); + } + } else if (this->stall_episode_) { + this->stall_episode_ = false; + SS_LOGI(TAG, "Source send recovered; %u buffered entries were dropped", + static_cast(this->stall_flushed_entries_)); + this->stall_flushed_entries_ = 0; + } + } + } + + // Close: stop accepting capture first so the tail below is finite + this->accepting_audio_.store(false, std::memory_order_release); + + // A final short chunk is allowed at stream end but not required (Sendspin spec, Source + // messages): a remainder the encoder cannot take is dropped rather than padded + if (this->chunk_filled_bytes_ > 0 && this->encoder_->can_encode(this->chunk_filled_bytes_)) { + this->send_chunk(conn); + this->chunk_filled_bytes_ = 0; + } + + // Sent from this thread so it is ordered after the last chunk by construction; best-effort, + // the connection may already be gone + conn->send_text_message(format_client_stream_end_message(), nullptr); + this->source_impl_->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STOPPED); + + this->flush_ring_to_live(); +} + +bool SourceTask::wait_for_time_sync(const std::shared_ptr& conn) { + // Asked of the bound connection directly: the stream must not open on the strength of a + // different connection's sync + while (!conn->is_time_synced()) { + if (!this->stream_still_open(conn)) { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(WAIT_FOR_TIME_SYNC_MS)); + } + return this->stream_still_open(conn); +} + +bool SourceTask::send_chunk(const std::shared_ptr& conn) { + // First-sample capture time minus encoder lookahead, converted with offset AND drift and + // never a playback/static delay term (Sendspin spec, Source messages -- timestamping) + const int64_t server_ts = conn->get_time_filter()->compute_server_time( + this->chunk_anchor_us_ - this->encoder_->lookahead_us()); + + uint8_t* staging = this->staging_.data(); + const size_t payload_len = this->encoder_->encode( + staging + SOURCE_WIRE_HEADER_SIZE, this->chunk_filled_bytes_, + staging + SOURCE_WIRE_HEADER_SIZE, this->staging_.size() - SOURCE_WIRE_HEADER_SIZE); + if (payload_len == 0) { + SS_LOGW(TAG, "Source encoder produced no payload; dropping chunk"); + return false; + } + + staging[0] = SENDSPIN_BINARY_SOURCE_AUDIO; + host_to_be64(server_ts, staging + 1); + + const SsErr err = conn->send_binary_message(staging, SOURCE_WIRE_HEADER_SIZE + payload_len, + this->send_complete_cb_); + + // Consume the completion unconditionally: the callback fires exactly once for EVERY call + // (connection.h contract), so skipping the wait on an error would leave a stale bit pacing + // the next send, and waiting every send out is what makes staging reuse and teardown safe + this->event_flags_.wait(SourceTaskBits::SOURCE_SEND_COMPLETE, false, true, UINT32_MAX); + return err == SsErr::OK && this->last_send_ok_.load(std::memory_order_acquire); +} + +size_t SourceTask::flush_ring_to_live() { + size_t flushed = 0; + if (this->current_entry_ != nullptr) { + this->capture_ring_->return_chunk(this->current_entry_); + this->current_entry_ = nullptr; + ++flushed; + } + this->entry_consumed_bytes_ = 0; + while (true) { + AudioRingBufferEntry* entry = this->capture_ring_->receive_chunk(0); + if (entry == nullptr) { + return flushed; + } + this->capture_ring_->return_chunk(entry); + ++flushed; + } +} + +bool SourceTask::stream_still_open(const std::shared_ptr& conn) const { + if ((this->event_flags_.get() & SourceTaskBits::SOURCE_COMMAND_STOP) != 0U) { + return false; + } + if (!this->stream_requested_.load(std::memory_order_acquire)) { + return false; + } + // Permission is per-connection: the stream ends as soon as its connection stops being + // current (drop or handoff), without waiting for the main loop's cleanup to latch the stop. + return this->connections_->current_shared() == conn; +} + +void SourceTask::stop() { + if (!this->task_thread_.joinable()) { + return; + } + this->event_flags_.set(SourceTaskBits::SOURCE_COMMAND_STOP); + this->task_thread_.join(); +} + +} // namespace sendspin diff --git a/src/source_task.h b/src/source_task.h new file mode 100644 index 0000000..c6f8434 --- /dev/null +++ b/src/source_task.h @@ -0,0 +1,244 @@ +// 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 source_task.h +/// @brief Background task that assembles captured audio into timestamped chunks and streams +/// them to the Sendspin server + +#pragma once + +#include "audio_ring_buffer.h" +#include "constants.h" +#include "platform/event_flags.h" +#include "platform/memory.h" +#include "sendspin/source_role.h" +#include "source_encoder.h" + +#include +#include +#include +#include + +namespace sendspin { + +class ConnectionManager; +class SendspinClient; +class SendspinConnection; + +/// @brief Event flag bits for source task lifecycle and command signaling (distinct names from +/// the sync task's EventGroupBits: both unscoped enums can be visible in one translation unit) +enum SourceTaskBits : uint16_t { + SOURCE_COMMAND_STOP = (1 << 0), // Signal task thread to exit + SOURCE_COMMAND_UPDATE = (1 << 1), // Desired streaming state changed; re-read the atomic + SOURCE_SEND_COMPLETE = (1 << 2), // A binary send's completion callback fired + SOURCE_TASK_RUNNING = (1 << 8), // Task is actively streaming to the server + SOURCE_TASK_STOPPED = (1 << 10), // Task thread has exited + SOURCE_TASK_IDLE = (1 << 12), // Task is idle, waiting for a start command +}; + +// ============================================================================ +// Chunk/timestamp bookkeeping (pure helpers) +// ============================================================================ + +/// @brief Bytes in one frame (one sample across all channels) of the given capture format +constexpr size_t source_bytes_per_frame(uint8_t channels, uint8_t bit_depth) { + return static_cast(channels) * (static_cast(bit_depth) / 8U); +} + +/// @brief Frames in `ms` milliseconds at sample_rate. 64-bit so the product cannot wrap on +/// 32-bit targets; SourceTask::init() caps the derived byte sizes before narrowing. +constexpr uint64_t source_ms_to_frames(uint32_t ms, uint32_t sample_rate) { + return static_cast(ms) * sample_rate / MS_PER_SECOND; +} + +/// @brief Duration in microseconds of frame_count frames at sample_rate +constexpr int64_t source_frames_to_us(uint64_t frame_count, uint32_t sample_rate) { + return static_cast(frame_count * US_PER_SECOND / sample_rate); +} + +/// @brief Capture time of the first not-yet-consumed sample in a ring entry: a chunk starting +/// mid-entry anchors on its own first sample (Sendspin spec, Source messages), not the entry's +constexpr int64_t source_entry_anchor_us(int64_t entry_capture_us, size_t consumed_bytes, + size_t bytes_per_frame, uint32_t sample_rate) { + return entry_capture_us + source_frames_to_us(consumed_bytes / bytes_per_frame, sample_rate); +} + +/** + * @brief Background task that drains the capture ring, assembles timestamped chunks, and sends + * them (plus the stream's client-stream/start and end messages, so wire ordering holds by + * construction) on the connection the stream was opened on + * + * The thread starts once and idles between streams (no create/destroy churn on embedded); the + * desired streaming state is a latest-wins atomic the task converges on. + */ +class SourceTask { +public: + SourceTask() = default; + ~SourceTask(); + + /// @brief Initializes event flags, the capture ring, and the chunk staging buffer + /// @param source_impl The owning SourceRole::Impl, used for config, events, and listener. + /// @param client The owning SendspinClient, used for shared services. + /// @param connections The client's connection manager (outlives this task); each stream is + /// bound to one connection via current_shared(). + /// @return true on success, false on allocation failure. + bool init(SourceRole::Impl* source_impl, SendspinClient* client, + ConnectionManager* connections); + + /// @brief Creates and starts the persistent source background thread + /// Call once after init(). The thread idles until signal_start(). + /// @param task_stack_in_psram Whether to allocate the task stack in PSRAM (ESP-IDF only). + /// @param priority Thread priority (ESP-IDF only). + /// @return true if thread started successfully, false otherwise. + bool start(bool task_stack_in_psram, unsigned priority); + + /// @brief Returns true if init() has been called successfully + /// @return true if the source task has been initialized, false otherwise. + bool is_initialized() const { + return this->event_flags_.is_created(); + } + + /// @brief Returns true if the task is opening or running a stream + /// @return true from stream bring-up (time-sync wait) until stream close, false when idle + /// or stopped. + bool is_running() const { + // Reading uncreated event flags is a null-handle crash on ESP + if (!this->is_initialized()) { + return false; + } + return (this->event_flags_.get() & SourceTaskBits::SOURCE_TASK_RUNNING) != 0U; + } + + /// @brief Requests the task to open the outbound stream. Non-blocking + /// Thread-safe: may be called from any context. + void signal_start(); + + /// @brief Requests the task to close the outbound stream. Non-blocking + /// Thread-safe: may be called from any context. + void signal_stop(); + + /// @brief Immediately closes the write_audio() gate without waiting for the task + /// + /// Cleanup uses this before publishing its synthetic STOPPED event so + /// on_streaming_stopped() can never fire while write_audio() still accepts; the task also + /// closes the gate itself on every stream exit (an extra store is harmless). + /// Thread-safe: may be called from any context. + void close_audio_gate() { + this->accepting_audio_.store(false, std::memory_order_release); + } + + /// @brief Writes captured audio into the capture ring + /// + /// Hot path: exactly one producer thread, non-blocking, non-allocating. Rejects writes + /// while the stream is not open and non-whole-frame writes (a partial frame would shift + /// every later sample across channels). + /// @param data Interleaved PCM in the configured format. + /// @param len Length in bytes; must be a whole number of frames. + /// @param capture_time_us Local-clock capture time of the first sample; 0 stamps now. + /// @return true if buffered; false when rejected or the ring is full (drop counted, warned + /// once per episode). + bool write_audio(const uint8_t* data, size_t len, int64_t capture_time_us); + +protected: + /// @brief Entry point for the persistent source background thread + /// @param params Pointer to the owning SourceTask instance. + static void thread_entry(void* params); + + /// @brief Outer task loop: idle -> bind -> stream -> idle + void run(); + + /// @brief Runs one stream on the given bound connection: waits for time sync, sends + /// client-stream/start, loops assembling and sending chunks, and closes with + /// client-stream/end. Returns when the desired state drops to stopped, the bound + /// connection stops being current, or the task is told to exit. + void stream(const std::shared_ptr& conn); + + /// @brief Waits until the bound connection's time filter has a measurement (Sendspin spec, + /// Source messages: a source must not stream before time sync converges) + /// @return true once synced; false if streaming was cancelled while waiting. + bool wait_for_time_sync(const std::shared_ptr& conn); + + /// @brief Encodes, stamps, and sends the currently assembled chunk on the bound connection + /// @param conn The bound connection. + /// @return true if the send completed successfully, false on any failure (the caller + /// treats a failure as a stall: the chunk is dropped and the ring flushed). + bool send_chunk(const std::shared_ptr& conn); + + /// @brief Returns the in-progress ring entry (if any) and discards everything buffered in + /// the capture ring, so streaming resumes from live capture (the spec's stall policy). + /// @return Number of ring entries discarded. + size_t flush_ring_to_live(); + + /// @brief True while the task should keep the current stream open: desired state still + /// streaming, task not told to exit, and the bound connection still current (streaming + /// permission is per-connection; a swap ends the stream). + bool stream_still_open(const std::shared_ptr& conn) const; + + /// @brief Signals the task to stop and waits for the thread to finish + void stop(); + + // Struct fields + EventFlags event_flags_; + /// Built once in init(); per-send copies stay in std::function's small-buffer storage (no + /// allocation). Must not call back into the connection (may run mid-teardown on the + /// transport's thread): it only records last_send_ok_ and sets SOURCE_SEND_COMPLETE. + std::function send_complete_cb_; + /// Wire chunk under assembly: [type byte][BE64 server-clock capture µs][payload] + /// (Sendspin spec, Source messages) + PlatformBuffer staging_; + std::thread task_thread_; + + // Pointer fields + SendspinClient* client_{nullptr}; + ConnectionManager* connections_{nullptr}; + /// Entry being consumed across chunk boundaries; returned once fully consumed or on flush + AudioRingBufferEntry* current_entry_{nullptr}; + std::unique_ptr capture_ring_; + std::unique_ptr encoder_; + SourceRole::Impl* source_impl_{nullptr}; + + // 64-bit fields + /// Local-clock capture time of the assembled chunk's first sample + int64_t chunk_anchor_us_{0}; + + // size_t fields + size_t bytes_per_frame_{0}; + size_t chunk_bytes_{0}; // Payload bytes per full chunk: chunk_frames * bytes_per_frame + /// Bytes of the current entry already consumed into chunks + size_t entry_consumed_bytes_{0}; + /// Payload bytes assembled into staging_ so far for the chunk in progress + size_t chunk_filled_bytes_{0}; + /// Ring entries dropped in the current task-side stall episode (failed sends) + size_t stall_flushed_entries_{0}; + + // 32-bit fields + /// Producer-thread-only count of writes dropped in the current overflow episode + uint32_t producer_dropped_writes_{0}; + + // 8-bit fields + /// write_audio() gate, open only between client-stream/start and stream close. A lone + /// atomic, not a ShadowSlot: single flag on a lock-free hot path (the Inbox bitmask trade) + std::atomic accepting_audio_{false}; + /// Latest-wins desired streaming state from the role's main-loop command latch + std::atomic stream_requested_{false}; + /// Result of the last binary send, written before SOURCE_SEND_COMPLETE is set + std::atomic last_send_ok_{false}; + /// Task-side stall episode (send failed); episode edges are the only log sites + bool stall_episode_{false}; + // Producer-thread-only episode flags for write_audio()'s throttled warnings + bool producer_drop_episode_{false}; + bool producer_frame_warned_{false}; +}; + +} // namespace sendspin diff --git a/src/sync_task.cpp b/src/sync_task.cpp index 0b318b2..6358713 100644 --- a/src/sync_task.cpp +++ b/src/sync_task.cpp @@ -42,9 +42,6 @@ static constexpr uint32_t INITIAL_SYNC_ZEROS_DURATION_MS = 25; static constexpr size_t SYNC_TASK_STACK_SIZE = 6192; // Opus uses more stack than FLAC -/// @brief Wait time (ms) between retries when time sync is not yet available -static constexpr uint32_t WAIT_FOR_TIME_SYNC_MS = 15U; - /// @brief Timeout (ms) for receiving the next encoded audio chunk from the ring buffer static constexpr uint32_t ENCODED_CHUNK_RECEIVE_TIMEOUT_MS = 15U; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ac5113c..42be53f 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -31,6 +31,12 @@ add_executable(sendspin_tests test_artwork_role.cpp ) +# The source suite is registered only when the role is compiled in, so a +# -DSENDSPIN_ENABLE_SOURCE=OFF test build still links. +if(SENDSPIN_ENABLE_SOURCE) + target_sources(sendspin_tests PRIVATE test_source_role.cpp) +endif() + # 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_source_role.cpp b/tests/test_source_role.cpp new file mode 100644 index 0000000..295c5e8 --- /dev/null +++ b/tests/test_source_role.cpp @@ -0,0 +1,363 @@ +// 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. + +// Source role tests. Two layers, mirroring the acceptance criteria: +// - Pure chunk/timestamp bookkeeping helpers from source_task.h, tested directly. +// - SourceRole::Impl-level tests (make_impl pattern from test_artwork_role.cpp): config +// validation, hello/state field building, the command latch's fail-closed path, and +// lifecycle callback pairing. + +#include "connection_manager.h" +#include "inbox.h" +#include "platform/time.h" +#include "protocol_messages.h" +#include "sendspin/client.h" +#include "sendspin/config.h" +#include "source_encoder.h" +#include "source_role_impl.h" +#include "source_task.h" +#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 + +namespace { + +// ============================================================================ +// Pure bookkeeping helpers (source_task.h) +// ============================================================================ + +// Defends the chunk_bytes_ derivation in SourceTask::init(): chunk_frames must be exactly +// chunk_duration_ms * sample_rate / 1000. +TEST(SourceBookkeeping, MsToFramesDerivesChunkFrames) { + EXPECT_EQ(source_ms_to_frames(25, 48000), 1200U); + EXPECT_EQ(source_ms_to_frames(5, 8000), 40U); + EXPECT_EQ(source_ms_to_frames(150, 44100), 6615U); + // Control: a duration too short to hold a frame truncates to zero (init() fails closed on + // this). + EXPECT_EQ(source_ms_to_frames(5, 100), 0U); + // 64-bit contract: the ms x rate intermediate here is 4.8e9 > 2^32, which 32-bit + // arithmetic (size_t on ESP32) would wrap to 505032 frames; init() additionally caps the + // derived byte sizes before narrowing. + EXPECT_EQ(source_ms_to_frames(100000, 48000), 4800000ULL); +} + +// Defends write_audio()'s whole-frame validation unit and the payload sizing. +TEST(SourceBookkeeping, BytesPerFrame) { + EXPECT_EQ(source_bytes_per_frame(2, 16), 4U); + EXPECT_EQ(source_bytes_per_frame(2, 24), 6U); // 24-bit = 3 packed bytes per sample + EXPECT_EQ(source_bytes_per_frame(1, 32), 4U); +} + +TEST(SourceBookkeeping, FramesToUs) { + EXPECT_EQ(source_frames_to_us(1200, 48000), 25000); + EXPECT_EQ(source_frames_to_us(48000, 48000), 1000000); + // Truncation, not rounding: 1 frame at 48 kHz is 20.83 µs. + EXPECT_EQ(source_frames_to_us(1, 48000), 20); +} + +// Defends the chunk-anchor bookkeeping in SourceTask::stream(): the wire timestamp anchors on +// the capture time of the chunk's FIRST sample, advanced past the frames of the entry already +// consumed by earlier chunks. +TEST(SourceBookkeeping, EntryAnchorAdvancesAcrossChunkBoundaries) { + constexpr uint32_t RATE = 48000; + constexpr size_t BPF = 4; + + // A chunk that starts at the top of an entry anchors on the entry timestamp itself. + EXPECT_EQ(source_entry_anchor_us(1000000, 0, BPF, RATE), 1000000); + + // Misaligned entry-vs-chunk sizes: entries of 100 frames, chunks of 64 frames. The third + // chunk starts 28 frames into the second entry (chunk 1 took frames 0-63 of entry 0, + // chunk 2 took frames 64-99 of entry 0 plus frames 0-27 of entry 1). + constexpr int64_t ENTRY1_TS = 5000000; + const int64_t anchor = source_entry_anchor_us(ENTRY1_TS, 28 * BPF, BPF, RATE); + EXPECT_EQ(anchor, ENTRY1_TS + source_frames_to_us(28, RATE)); + + // Consuming a whole 100-frame entry advances its anchor by exactly its duration. + EXPECT_EQ(source_entry_anchor_us(ENTRY1_TS, 100 * BPF, BPF, RATE), + ENTRY1_TS + source_frames_to_us(100, RATE)); +} + +// Defends every branch of PcmPassthroughEncoder::encode() (source_encoder.h): the aliased +// in == out shape the task uses, the separate-buffer shape the interface contract also +// requires, and the capacity guard. +TEST(SourceBookkeeping, PcmPassthroughEncoderContract) { + PcmPassthroughEncoder encoder; + uint8_t buffer[8] = {1, 2, 3, 4, 5, 6, 7, 8}; + + // Aliased (the production shape): returns the length, bytes untouched. + EXPECT_EQ(encoder.encode(buffer, sizeof(buffer), buffer, sizeof(buffer)), sizeof(buffer)); + EXPECT_EQ(buffer[0], 1); + EXPECT_EQ(buffer[7], 8); + + // Separate buffers: the payload is copied. + uint8_t out[8] = {0}; + EXPECT_EQ(encoder.encode(buffer, sizeof(buffer), out, sizeof(out)), sizeof(buffer)); + EXPECT_EQ(0, memcmp(buffer, out, sizeof(buffer))); + + // Capacity guard: input larger than the output area encodes nothing. + uint8_t small[4] = {0}; + EXPECT_EQ(encoder.encode(buffer, sizeof(buffer), small, sizeof(small)), 0U); + + EXPECT_EQ(encoder.lookahead_us(), 0); + + // can_encode(): PCM takes any nonempty chunk, so a stream-end remainder always goes out. + EXPECT_TRUE(encoder.can_encode(6)); + EXPECT_FALSE(encoder.can_encode(0)); +} + +// ============================================================================ +// Impl-level harness +// ============================================================================ + +/// Records the source lifecycle callbacks. Callbacks fire on the pumping/test thread, so plain +/// counters suffice. +class RecordingSourceListener : public SourceRoleListener { +public: + void on_streaming_started() override { + ++this->started; + } + void on_streaming_stopped() override { + ++this->stopped; + } + + int started{0}; + int stopped{0}; +}; + +// A real, never-started SendspinClient plus a bound SourceRole::Impl. Heap-allocated with +// program lifetime (static deques, mirroring make_impl() in test_artwork_role.cpp): the Impl +// keeps raw SendspinClient* and ConnectionManager* pointers, so both must outlive it. The +// standalone ConnectionManager never holds a connection, so current() stays null -- exactly the +// stale-connection shape the fail-closed latch tests need. +std::unique_ptr make_impl(SourceRoleConfig config) { + static std::deque clients; + static std::deque managers; + static std::deque inboxes; + + clients.emplace_back(SendspinClientConfig{}); + managers.emplace_back(&clients.back()); + inboxes.emplace_back(); + + auto impl = std::make_unique(config, &clients.back()); + impl->attach_inbox(inboxes.back()); + impl->connection_manager = &managers.back(); + return impl; +} + +/// Replays pending SOURCE_STREAM ring events into the impl and drains it, mirroring the +/// dispatch order in SendspinClient::loop() (ring events first, then the role drain). +void pump_ring_and_drain(SourceRole::Impl& impl) { + InboxEvent events[Inbox::EVENT_CAPACITY]; + size_t count = impl.inbox->take_events(events, Inbox::EVENT_CAPACITY); + for (size_t i = 0; i < count; ++i) { + if (events[i].type == InboxEventType::SOURCE_STREAM) { + impl.on_stream_ring_event(static_cast(events[i].code)); + } + } + impl.drain_events(); +} + +bool advertises_source(SourceRole::Impl& impl) { + ClientHelloMessage msg; + impl.build_hello_fields(msg); + return std::find(msg.supported_roles.begin(), msg.supported_roles.end(), + SendspinRole::SOURCE) != msg.supported_roles.end(); +} + +// ============================================================================ +// Config validation (rejected configs leave the role inert, never clamped) +// ============================================================================ + +// Defends validate_config()'s chunk-duration bounds in source_role.cpp: the spec's [5, 150] ms +// window rejects outside values instead of clamping them. +TEST(SourceConfigValidation, ChunkDurationBounds) { + auto make_with_chunk = [](uint32_t ms) { + SourceRoleConfig config; + config.chunk_duration_ms = ms; + return make_impl(config); + }; + EXPECT_FALSE(advertises_source(*make_with_chunk(4))); + EXPECT_FALSE(advertises_source(*make_with_chunk(151))); + // Control: the bound values themselves and the default are accepted. + EXPECT_TRUE(advertises_source(*make_with_chunk(5))); + EXPECT_TRUE(advertises_source(*make_with_chunk(25))); + EXPECT_TRUE(advertises_source(*make_with_chunk(150))); +} + +// Defends validate_config()'s format checks: zero rate/channels and unsupported bit depths +// reject the whole config. +TEST(SourceConfigValidation, FormatFields) { + { + SourceRoleConfig config; + config.sample_rate = 0; + EXPECT_FALSE(advertises_source(*make_impl(config))); + } + { + SourceRoleConfig config; + config.channels = 0; + EXPECT_FALSE(advertises_source(*make_impl(config))); + } + for (uint8_t depth : {uint8_t{8}, uint8_t{12}}) { + SourceRoleConfig config; + config.bit_depth = depth; + EXPECT_FALSE(advertises_source(*make_impl(config))); + } + // Control: every supported depth is accepted. + for (uint8_t depth : {uint8_t{16}, uint8_t{24}, uint8_t{32}}) { + SourceRoleConfig config; + config.bit_depth = depth; + EXPECT_TRUE(advertises_source(*make_impl(config))); + } +} + +// Defends build_hello_fields(): a valid role advertises source@v1 with the support object +// carrying the configured line_sense flag. +TEST(SourceHello, AdvertisesSupportObject) { + SourceRoleConfig config; + config.line_sense = true; + auto impl = make_impl(config); + + ClientHelloMessage msg; + impl->build_hello_fields(msg); + ASSERT_TRUE(advertises_source(*impl)); + ASSERT_TRUE(msg.source_v1_support.has_value()); + EXPECT_TRUE(msg.source_v1_support->line_sense); +} + +// Defends build_state_fields() and set_signal(): the source state object is always present for +// a valid role, carries signal only after set_signal() with line_sense configured, and an +// invalid config contributes nothing. +TEST(SourceState, SignalOnlyWithLineSense) { + { + auto impl = make_impl(SourceRoleConfig{}); // line_sense defaults off + impl->set_signal(SourceSignal::PRESENT); // ignored with a warning + ClientStateMessage msg; + impl->build_state_fields(msg); + ASSERT_TRUE(msg.source.has_value()); + EXPECT_FALSE(msg.source->signal.has_value()); + } + { + SourceRoleConfig config; + config.line_sense = true; + auto impl = make_impl(config); + { + ClientStateMessage msg; + impl->build_state_fields(msg); + ASSERT_TRUE(msg.source.has_value()); + EXPECT_FALSE(msg.source->signal.has_value()); // no signal reported yet + } + impl->set_signal(SourceSignal::PRESENT); + ClientStateMessage msg; + impl->build_state_fields(msg); + ASSERT_TRUE(msg.source.has_value()); + ASSERT_TRUE(msg.source->signal.has_value()); + EXPECT_EQ(msg.source->signal.value(), SourceSignal::PRESENT); + } + { + SourceRoleConfig config; + config.sample_rate = 0; // invalid + auto impl = make_impl(config); + ClientStateMessage msg; + impl->build_state_fields(msg); + EXPECT_FALSE(msg.source.has_value()); + } +} + +// ============================================================================ +// Command latch (fail closed) and lifecycle pairing +// ============================================================================ + +// Defends the instance-id check in drain_events(): a start command whose connection is no +// longer current (here: no connection at all) must be discarded, never latched -- streaming +// permission is per-connection. +TEST(SourceCommandLatch, StaleInstanceStartDiscarded) { + auto impl = make_impl(SourceRoleConfig{}); + ASSERT_TRUE(impl->start()); + + impl->handle_server_command(SourceCommand::START, 7); + impl->drain_events(); + + EXPECT_FALSE(impl->streaming_desired); + EXPECT_FALSE(impl->task->is_running()); +} + +// Defends cleanup(): the latch resets to stopped unconditionally and the unconditional +// STOPPED event fires on_streaming_stopped() exactly once, paired with the earlier start. +TEST(SourceCommandLatch, CleanupResetsLatchAndPairsStop) { + auto impl = make_impl(SourceRoleConfig{}); + RecordingSourceListener listener; + impl->listener = &listener; + + // Simulate an open stream as the task would report it. + impl->streaming_desired = true; + impl->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STARTED); + pump_ring_and_drain(*impl); + ASSERT_EQ(listener.started, 1); + ASSERT_TRUE(impl->streaming_active); + + impl->cleanup(); + EXPECT_FALSE(impl->streaming_desired); + pump_ring_and_drain(*impl); + EXPECT_EQ(listener.stopped, 1); + EXPECT_FALSE(impl->streaming_active); + + // A second teardown (double-teardown tick) must not fire an unpaired second callback. + impl->cleanup(); + pump_ring_and_drain(*impl); + EXPECT_EQ(listener.stopped, 1); +} + +// Defends the streaming_active gate in drain_events(): duplicate STARTED/STOPPED ring events +// collapse to one callback each, keeping the listener pairing 1:1. +TEST(SourceCommandLatch, DuplicateLifecycleEventsCollapse) { + auto impl = make_impl(SourceRoleConfig{}); + RecordingSourceListener listener; + impl->listener = &listener; + + impl->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STARTED); + impl->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STARTED); + pump_ring_and_drain(*impl); + EXPECT_EQ(listener.started, 1); + + impl->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STOPPED); + impl->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STOPPED); + pump_ring_and_drain(*impl); + EXPECT_EQ(listener.stopped, 1); +} + +// Defends the accepting gate in SourceTask::write_audio(): with no open stream every write is +// rejected, whole frames or not. +TEST(SourceWriteAudio, RejectedWhenNotStreaming) { + auto impl = make_impl(SourceRoleConfig{}); + ASSERT_TRUE(impl->start()); + + const uint8_t frame[4] = {1, 2, 3, 4}; + EXPECT_FALSE(impl->write_audio(frame, sizeof(frame), 0)); +} + +} // namespace