From dbbd93e6275bc0dfc9ee5fca42066bba4e112da2 Mon Sep 17 00:00:00 2001 From: marcelveldt Date: Wed, 16 Sep 2026 09:57:30 +0200 Subject: [PATCH 1/2] Add announcement role (announcement@v1) --- CMakeLists.txt | 18 + Kconfig | 7 + cmake/sources.cmake | 8 + include/sendspin/announcement_role.h | 172 +++++++++ include/sendspin/client.h | 23 ++ include/sendspin/config.h | 30 ++ src/announcement_role.cpp | 395 +++++++++++++++++++ src/announcement_role_impl.h | 116 ++++++ src/announcement_task.cpp | 416 ++++++++++++++++++++ src/announcement_task.h | 172 +++++++++ src/client.cpp | 92 ++++- src/inbox.h | 17 +- src/protocol.cpp | 75 ++++ src/protocol_messages.h | 57 ++- tests/CMakeLists.txt | 1 + tests/test_announcement_role.cpp | 553 +++++++++++++++++++++++++++ tests/test_connection_lifecycle.cpp | 72 ++++ 17 files changed, 2200 insertions(+), 24 deletions(-) create mode 100644 include/sendspin/announcement_role.h create mode 100644 src/announcement_role.cpp create mode 100644 src/announcement_role_impl.h create mode 100644 src/announcement_task.cpp create mode 100644 src/announcement_task.h create mode 100644 tests/test_announcement_role.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index e79d8f8f..2ae77259 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -53,6 +53,12 @@ if(ESP_IDF_BUILD) list(APPEND SENDSPIN_COMPILE_DEFS SENDSPIN_ENABLE_PLAYER) endif() + if(CONFIG_SENDSPIN_ENABLE_ANNOUNCEMENT) + # Kconfig enforces the player dependency (ring buffer + decoder sources) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_ANNOUNCEMENT_SOURCES}) + list(APPEND SENDSPIN_COMPILE_DEFS SENDSPIN_ENABLE_ANNOUNCEMENT) + endif() + if(CONFIG_SENDSPIN_ENABLE_CONTROLLER) list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_CONTROLLER_SOURCES}) list(APPEND SENDSPIN_COMPILE_DEFS SENDSPIN_ENABLE_CONTROLLER) @@ -109,6 +115,12 @@ 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_ANNOUNCEMENT "Enable announcement role (requires player)" ON) + + if(SENDSPIN_ENABLE_ANNOUNCEMENT AND NOT SENDSPIN_ENABLE_PLAYER) + # The announcement role reuses the player's ring buffer and decoder sources + message(FATAL_ERROR "SENDSPIN_ENABLE_ANNOUNCEMENT requires SENDSPIN_ENABLE_PLAYER") + endif() include(${CMAKE_CURRENT_SOURCE_DIR}/cmake/host.cmake) @@ -136,6 +148,9 @@ else() if(SENDSPIN_ENABLE_VISUALIZER) list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_VISUALIZER_SOURCES}) endif() + if(SENDSPIN_ENABLE_ANNOUNCEMENT) + list(APPEND SENDSPIN_ALL_SOURCES ${SENDSPIN_ANNOUNCEMENT_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 +174,9 @@ else() if(SENDSPIN_ENABLE_VISUALIZER) target_compile_definitions(sendspin PUBLIC SENDSPIN_ENABLE_VISUALIZER) endif() + if(SENDSPIN_ENABLE_ANNOUNCEMENT) + target_compile_definitions(sendspin PUBLIC SENDSPIN_ENABLE_ANNOUNCEMENT) + endif() # Host build options option(ENABLE_WERROR "Treat warnings as errors" OFF) diff --git a/Kconfig b/Kconfig index 6c724682..88a7f633 100644 --- a/Kconfig +++ b/Kconfig @@ -35,4 +35,11 @@ menu "sendspin-cpp" bool "Enable visualizer role" default y + config SENDSPIN_ENABLE_ANNOUNCEMENT + bool "Enable announcement role (draft)" + # Reuses the player role's ring buffer and decoder sources + depends on SENDSPIN_ENABLE_PLAYER + # Draft role (announcement@v1 spec proposal); opt-in until the spec lands + default n + endmenu diff --git a/cmake/sources.cmake b/cmake/sources.cmake index d32f30c5..15567214 100644 --- a/cmake/sources.cmake +++ b/cmake/sources.cmake @@ -36,6 +36,14 @@ function(sendspin_get_sources BASE_DIR) PARENT_SCOPE ) + # Requires SENDSPIN_PLAYER_SOURCES: reuses the player's ring buffer and decoder + set(SENDSPIN_ANNOUNCEMENT_SOURCES + ${BASE_DIR}/src/announcement_role.cpp + ${BASE_DIR}/src/announcement_task.cpp + + PARENT_SCOPE + ) + set(SENDSPIN_CONTROLLER_SOURCES ${BASE_DIR}/src/controller_role.cpp diff --git a/include/sendspin/announcement_role.h b/include/sendspin/announcement_role.h new file mode 100644 index 00000000..9e28fc19 --- /dev/null +++ b/include/sendspin/announcement_role.h @@ -0,0 +1,172 @@ +// 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 announcement_role.h +/// @brief Announcement role that decodes short per-client audio clips (TTS, chimes, alerts) +/// concurrently with the media stream, with ducking hints for the embedder + +#pragma once + +#include "sendspin/config.h" +#include "sendspin/player_role.h" // ServerPlayerStreamObject (shared codec-parameter shape) +#include "sendspin/types.h" + +#include +#include +#include +#include + +namespace sendspin { + +class SendspinClient; + +// ============================================================================ +// Announcement types +// ============================================================================ + +/// @brief Announcement stream parameters sent by the server in stream/start messages +/// +/// The codec parameters share the player stream-object shape; the additional fields carry the +/// per-announcement ducking policy and optional output level. +struct ServerAnnouncementStreamObject { + /// @brief Default ducking ramp duration in milliseconds, used when the server omits the field + static constexpr uint16_t DEFAULT_DUCK_RAMP_MS = 100U; + + ServerPlayerStreamObject format{}; + + /// Server clock time in microseconds when announcement output should begin. The client + /// translates it via clock synchronization and starts at that time, or as soon as possible + /// if it has already passed. The same value on several clients gives a coordinated + /// multi-speaker start. Required. + int64_t start_timestamp{0}; + + /// Reduction in decibel (0-50) to apply to this client's own media output while the + /// announcement stream is active. 0 means no ducking. + uint8_t media_duck_db{0}; + + /// Ramp duration in milliseconds (0-2000) for both applying and releasing the ducking. + uint16_t duck_ramp_ms{DEFAULT_DUCK_RAMP_MS}; + + /// When set, the announcement should be rendered at the loudness that master volume + /// `volume` (0-100) would produce, regardless of the current master volume. When absent, + /// the announcement follows the current master volume. + std::optional volume{}; + + /// When true, a muted client still renders the announcement while its media stays muted. + bool override_mute{false}; + + bool is_complete() const { + return this->format.is_complete(); + } +}; + +/// @brief Listener for announcement role events +/// +/// THREAD SAFETY: on_announcement_write() fires on the announcement task's background thread. +/// Implementations must be thread-safe for this method. on_announcement_start() and +/// on_announcement_end() fire on the main loop thread via drain_events(). The listener must +/// outlive the role. +class AnnouncementRoleListener { +public: + virtual ~AnnouncementRoleListener() = default; + + /// @brief Writes decoded announcement PCM audio to the platform's announcement output + /// + /// Fires on the announcement task's background thread. May block up to timeout_ms; the + /// blocking write is what paces the decode loop against the sink. + /// @param data Pointer to the decoded PCM audio data + /// @param length Number of bytes to write; always a whole number of PCM frames + /// @param timeout_ms Maximum time to wait for the write to complete + /// @return Number of bytes actually written; partial writes must be whole PCM frames + virtual size_t on_announcement_write(uint8_t* data, size_t length, uint32_t timeout_ms) = 0; + + /// @brief Called when an announcement stream starts. Fires on the main loop thread + /// + /// The embedder applies the ducking policy here (e.g. duck the media pipeline by + /// `params.media_duck_db` over `params.duck_ramp_ms`) and honors `params.override_mute` and + /// `params.volume`. Also fires when the server re-sends `stream/start` to update the active + /// announcement's config (duck level, volume, or override_mute): the embedder re-applies the + /// updated policy, ramping a new `media_duck_db` from the current gain, without any + /// intervening on_announcement_end(). + virtual void on_announcement_start(const ServerAnnouncementStreamObject& /*params*/) {} + + /// @brief Called when the announcement stream ends (normally, aborted, or on transport + /// loss). Fires on the main loop thread. The embedder releases the ducking here + virtual void on_announcement_end() {} +}; + +/** + * @brief Announcement role that decodes short per-client audio clips next to the media stream + * + * Owns an AnnouncementTask that runs on a background thread. Encoded announcement chunks + * arrive from the WebSocket network thread, are written into a dedicated ring buffer, decoded, + * and delivered to the platform through AnnouncementRoleListener::on_announcement_write(). + * Unlike the player role there is no sample-accurate sync machinery: announcements are + * per-client, start at (or as soon as possible after) the stream's single `start_timestamp` - + * which is also what aligns a coordinated multi-speaker announcement scheduled by the server - + * and are then paced by the sink. + * + * Usage: + * 1. Implement AnnouncementRoleListener with at minimum on_announcement_write() + * 2. Add the role to the client via SendspinClient::add_announcement() + * 3. Call set_listener() with your listener implementation + * + * @code + * struct MyAnnouncementListener : AnnouncementRoleListener { + * size_t on_announcement_write(uint8_t* data, size_t len, uint32_t timeout_ms) override { + * return announcement_output.write(data, len, timeout_ms); + * } + * void on_announcement_start(const ServerAnnouncementStreamObject& params) override { + * media_mixer.apply_ducking(params.media_duck_db, params.duck_ramp_ms); + * } + * void on_announcement_end() override { media_mixer.apply_ducking(0, ramp_ms); } + * }; + * + * MyAnnouncementListener listener; + * AnnouncementRoleConfig config; + * config.audio_formats = {{SendspinCodecFormat::PCM, 1, 48000, 16}}; + * auto& announcement = client.add_announcement(config); + * announcement.set_listener(&listener); + * @endcode + */ +class AnnouncementRole { + friend class SendspinClient; + +public: + struct Impl; + + AnnouncementRole(AnnouncementRoleConfig config, SendspinClient* client); + ~AnnouncementRole(); + + /// @brief Sets the listener for announcement events + /// @param listener Pointer to the listener implementation; must outlive this role + void set_listener(AnnouncementRoleListener* listener); + + // ======================================== + // Queries + // ======================================== + + /// @brief Returns a reference to the current announcement stream parameters + /// @return Const reference to the active announcement stream parameters. + const ServerAnnouncementStreamObject& get_current_stream_params() const; + + /// @brief Returns true if announcement audio is currently being output + /// @return true while the announcement pipeline is playing, false otherwise. + bool is_playing() const; + +private: + std::unique_ptr impl_; +}; + +} // namespace sendspin diff --git a/include/sendspin/client.h b/include/sendspin/client.h index e192c79f..d25190f1 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -32,6 +32,9 @@ namespace sendspin { // Forward declarations for enabled roles +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT +class AnnouncementRole; +#endif #ifdef SENDSPIN_ENABLE_ARTWORK class ArtworkRole; #endif @@ -312,10 +315,27 @@ class SendspinClient { VisualizerRole& add_visualizer(VisualizerRoleConfig config); #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + /// @brief Adds the announcement role. Returns a reference for setting callbacks + AnnouncementRole& add_announcement(AnnouncementRoleConfig config); +#endif + // ======================================== // Role access (nullptr if not added) // ======================================== +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + /// @brief Returns the announcement role, or nullptr if not added + /// @return Pointer to the announcement role, or nullptr + AnnouncementRole* announcement() { + return this->announcement_.get(); + } + /// @brief Returns the announcement role (const), or nullptr if not added + /// @return Const pointer to the announcement role, or nullptr + const AnnouncementRole* announcement() const { + return this->announcement_.get(); + } +#endif #ifdef SENDSPIN_ENABLE_ARTWORK /// @brief Returns the artwork role, or nullptr if not added /// @return Pointer to the artwork role, or nullptr @@ -538,6 +558,9 @@ class SendspinClient { GroupUpdateObject group_state_{}; // Pointer fields +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + std::unique_ptr announcement_; +#endif #ifdef SENDSPIN_ENABLE_ARTWORK std::unique_ptr artwork_; #endif diff --git a/include/sendspin/config.h b/include/sendspin/config.h index c77bf605..4f15164a 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -162,6 +162,36 @@ struct PlayerRoleConfig { MemoryLocation decode_buffer_location{MemoryLocation::PREFER_EXTERNAL}; }; +// ============================================================================ +// Announcement config types +// ============================================================================ + +/// @brief Configuration for the announcement role +struct AnnouncementRoleConfig { + /// @brief Default encoded announcement buffer size (~128 KB; announcements are short clips) + static constexpr size_t DEFAULT_AUDIO_BUFFER_CAPACITY = 131072U; + + /// Announcement formats advertised to the server, in priority order. These are decoded on a + /// second concurrent pipeline next to the media stream, so prefer inexpensive formats + /// (mono, 16-bit, modest sample rates). + std::vector audio_formats{}; + size_t audio_buffer_capacity{DEFAULT_AUDIO_BUFFER_CAPACITY}; + + bool psram_stack{false}; ///< Allocate announcement task stack in PSRAM (ESP-IDF only) + + /// @brief Default FreeRTOS priority for the announcement decode task (ESP-IDF only). + /// Below the httpd server and media sync tasks so a playing announcement can never starve + /// the synchronized media pipeline. + static constexpr unsigned DEFAULT_ANNOUNCEMENT_TASK_PRIORITY = 4U; + + unsigned priority{DEFAULT_ANNOUNCEMENT_TASK_PRIORITY}; ///< FreeRTOS priority for the + ///< announcement task (ESP-IDF only) + + /// @brief Memory placement for the decode buffer (ESP-IDF only; ignored on host). + /// Defaults to PREFER_EXTERNAL (SPIRAM). + MemoryLocation decode_buffer_location{MemoryLocation::PREFER_EXTERNAL}; +}; + // ============================================================================ // Artwork config types // ============================================================================ diff --git a/src/announcement_role.cpp b/src/announcement_role.cpp new file mode 100644 index 00000000..d5f788e9 --- /dev/null +++ b/src/announcement_role.cpp @@ -0,0 +1,395 @@ +// 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 "announcement_role_impl.h" +#include "audio_types.h" +#include "platform/base64.h" +#include "platform/compiler.h" +#include "platform/logging.h" +#include "protocol_messages.h" +#include "sendspin/client.h" + +static const char* const TAG = "sendspin.announcement"; + +static constexpr uint32_t HEADER_SEND_TIMEOUT_MS = 100U; +// Denominator for the advertised buffer capacity fraction: advertises (N-1)/N of capacity, +// matching the player role's ring-buffer metadata headroom. +static constexpr size_t ANNOUNCEMENT_BUFFER_ADVERTISE_DENOMINATOR = 5; + +namespace sendspin { + +// ============================================================================ +// Helpers +// ============================================================================ + +/// @brief Decodes a base64-encoded string into a byte vector. +static std::vector base64_decode(const std::string& input) { + size_t output_len = 0; + platform_base64_decode(nullptr, 0, &output_len, + reinterpret_cast(input.data()), input.size()); + + std::vector output(output_len); + int ret = + platform_base64_decode(output.data(), output.size(), &output_len, + reinterpret_cast(input.data()), input.size()); + if (ret != 0) { + SS_LOGW(TAG, "base64 decode failed: %d", ret); + return {}; + } + output.resize(output_len); + return output; +} + +// ============================================================================ +// Impl constructor / destructor +// ============================================================================ + +AnnouncementRole::Impl::Impl(AnnouncementRoleConfig config, SendspinClient* client) + : config(std::move(config)), + client(client), + event_state(std::make_unique()), + task(std::make_unique()) {} + +AnnouncementRole::Impl::~Impl() { + // Stop the announcement task thread first, before destroying other members: the task thread + // reads the listener pointer and the ring buffer until it has fully stopped. + this->task.reset(); +} + +// ============================================================================ +// AnnouncementRole forwarding (public API -> Impl) +// ============================================================================ + +AnnouncementRole::AnnouncementRole(AnnouncementRoleConfig config, SendspinClient* client) + : impl_(std::make_unique(std::move(config), client)) {} + +AnnouncementRole::~AnnouncementRole() = default; + +void AnnouncementRole::set_listener(AnnouncementRoleListener* listener) { + this->impl_->listener = listener; +} + +const ServerAnnouncementStreamObject& AnnouncementRole::get_current_stream_params() const { + return this->impl_->current_stream_params; +} + +bool AnnouncementRole::is_playing() const { + return this->impl_->announcement_playing; +} + +// ============================================================================ +// Impl: Internal integration methods +// ============================================================================ + +void AnnouncementRole::Impl::attach_inbox(Inbox& inbox) { + this->inbox = &inbox; + this->event_state->stream_params_slot.bind(inbox, INBOX_TOPIC_ANNOUNCEMENT_STREAM_PARAMS); +} + +bool AnnouncementRole::Impl::start() { + if (!this->config.audio_formats.empty() && this->listener && !this->task->is_initialized()) { + if (!this->task->init(this, this->client, this->config.audio_buffer_capacity)) { + SS_LOGE(TAG, "Failed to initialize announcement task"); + return false; + } + if (!this->task->start(this->config.psram_stack, this->config.priority)) { + SS_LOGE(TAG, "Failed to start announcement task thread"); + return false; + } + } + return true; +} + +void AnnouncementRole::Impl::build_hello_fields(ClientHelloMessage& msg) { + if (this->config.audio_formats.empty()) { + return; + } + + msg.supported_roles.push_back(SendspinRole::ANNOUNCEMENT); + + // Advertise 80% of the buffer capacity to account for ring buffer metadata overhead, + // matching the player role's convention + AnnouncementSupportObject announcement_support = { + .supported_formats = this->config.audio_formats, + .buffer_capacity = this->config.audio_buffer_capacity * + (ANNOUNCEMENT_BUFFER_ADVERTISE_DENOMINATOR - 1) / + ANNOUNCEMENT_BUFFER_ADVERTISE_DENOMINATOR, + }; + msg.announcement_v1_support = announcement_support; +} + +void AnnouncementRole::Impl::build_state_fields(ClientStateMessage& msg) const { + if (this->config.audio_formats.empty()) { + return; + } + + ClientAnnouncementStateObject announcement_state{}; + announcement_state.playing = this->announcement_playing; + msg.announcement = announcement_state; +} + +SS_HOT void AnnouncementRole::Impl::handle_binary(const uint8_t* data, size_t len) const { + if (this->config.audio_formats.empty()) { + return; + } + // Announcement chunks are untimed: the whole payload (after the stripped type byte) is one + // encoded audio frame. The stream's single start time travels with the codec header, not + // the chunks. Chunks are not sync-critical and must not be dropped for being late, so a full + // ring buffer is the only failure here (bounded by the advertised buffer_capacity). + if (!this->task->write_audio_chunk(data, len, 0, CHUNK_TYPE_ENCODED_AUDIO, 0)) { + SS_LOGW(TAG, "Failed to buffer announcement chunk"); + } +} + +void AnnouncementRole::Impl::handle_stream_start( + const ServerAnnouncementStreamObject& announcement_obj) const { + if (this->config.audio_formats.empty()) { + // No announcement formats configured; just defer the stream start callback + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_START); + return; + } + + // A stream/start arriving while the announcement is already playing is a configuration update + // (duck level, volume, or override_mute), not a new clip: update the params in place and let + // the main thread re-apply the policy via on_announcement_start, without a codec header, a + // buffer clear, or a task restart. Replacing a clip is done by the server with stream/end + // then a fresh stream/start. + if (this->task->is_running()) { + this->event_state->stream_params_slot.write(announcement_obj); + this->enqueue_stream_event(AnnouncementStreamCallbackType::CONFIG_UPDATE); + return; + } + + bool header_sent = false; + const ServerPlayerStreamObject& format = announcement_obj.format; + + // The codec header chunk carries the stream's single start_timestamp: it is the first chunk + // the task consumes, so the task reads the scheduled start time from it before any audio. + const int64_t start_timestamp = announcement_obj.start_timestamp; + + if (!format.bit_depth.has_value() || !format.channels.has_value() || + !format.sample_rate.has_value() || !format.codec.has_value()) { + SS_LOGE(TAG, "Announcement stream start missing required audio parameters"); + } else { + auto codec = format.codec.value(); + + if ((codec == SendspinCodecFormat::PCM) || (codec == SendspinCodecFormat::OPUS)) { + DummyHeader header{}; + header.sample_rate = format.sample_rate.value(); + header.bits_per_sample = format.bit_depth.value(); + header.channels = format.channels.value(); + + ChunkType chunk_type = (codec == SendspinCodecFormat::PCM) + ? CHUNK_TYPE_PCM_DUMMY_HEADER + : CHUNK_TYPE_OPUS_DUMMY_HEADER; + + header_sent = this->task->write_audio_chunk(reinterpret_cast(&header), + sizeof(DummyHeader), start_timestamp, + chunk_type, HEADER_SEND_TIMEOUT_MS); + if (!header_sent) { + SS_LOGE(TAG, "Failed to send announcement codec header"); + } + } else if (codec == SendspinCodecFormat::FLAC) { + if (!format.codec_header.has_value()) { + SS_LOGE(TAG, "FLAC codec header missing"); + } else { + std::vector flac_header = base64_decode(format.codec_header.value()); + header_sent = this->task->write_audio_chunk(flac_header.data(), flac_header.size(), + start_timestamp, CHUNK_TYPE_FLAC_HEADER, + HEADER_SEND_TIMEOUT_MS); + if (!header_sent) { + SS_LOGE(TAG, "Failed to send announcement codec header"); + } + } + } else { + SS_LOGE(TAG, "Unsupported announcement codec: %d", static_cast(codec)); + } + } + + if (!header_sent) { + this->task->signal_stream_end(); + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_END); + return; + } + + // Write stream params to the inbox slot for the main thread, then signal. Same two-step + // write/push pattern (and the same benign cleanup race) as the player role: a concurrent + // cleanup() may reset the slot between the two, in which case the drained STREAM_START + // finds the slot empty and keeps the prior params, with the teardown's STREAM_END queued + // right behind it. + this->event_state->stream_params_slot.write(announcement_obj); + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_START); +} + +void AnnouncementRole::Impl::handle_stream_end() const { + // A server stream/end is a graceful drain: the task plays out any buffered announcement audio + // before ending. The duck release (on_announcement_end) and the idle report are deferred to + // the task's OUTPUT_FINISHED event once the tail has played, so no STREAM_END event is queued + // here - queuing one would release the duck immediately and cut the tail off. + this->task->signal_stream_drain(); +} + +void AnnouncementRole::Impl::on_stream_ring_event(AnnouncementStreamCallbackType event) { + this->pending_events.push_back(event); +} + +void AnnouncementRole::Impl::drain_events() { + if (this->pending_events.empty()) { + return; + } + + bool state_changed = false; + size_t processed = 0; + bool teardown_reentered = false; + + // Indexed with a fresh size() check per iteration (not a range-for): the listener callbacks + // below may re-enter connection teardown, whose cleanup() clears this vector mid-loop (same + // pattern as the player role). + // NOLINTNEXTLINE(modernize-loop-convert): body mutates the vector, see above + for (size_t idx = 0; idx < this->pending_events.size(); ++idx) { + const AnnouncementStreamCallbackType event = this->pending_events[idx]; + + switch (event) { + case AnnouncementStreamCallbackType::STREAM_START: { + ServerAnnouncementStreamObject stream_params; + if (this->event_state->stream_params_slot.take(stream_params)) { + this->current_stream_params = std::move(stream_params); + } + // Mark the stream active before invoking the listener so the STREAM_END that a + // re-entrant cleanup() enqueues still delivers a paired on_announcement_end(). + this->stream_active = true; + if (this->listener) { + const uint32_t generation = this->cleanup_generation; + this->listener->on_announcement_start(this->current_stream_params); + if (this->cleanup_generation != generation) { + teardown_reentered = true; + break; + } + } + this->task->signal_stream_start(); + break; + } + case AnnouncementStreamCallbackType::STREAM_END: { + // Immediate, non-graceful end (connection teardown, or a stream that failed to + // start): release the ducking now. A graceful server stream/end does not reach + // here - it releases via OUTPUT_FINISHED after the buffered tail has played. + if (this->listener && this->stream_active) { + this->listener->on_announcement_end(); + } + this->stream_active = false; + if (this->announcement_playing) { + this->announcement_playing = false; + state_changed = true; + } + break; + } + case AnnouncementStreamCallbackType::CONFIG_UPDATE: { + // Re-sent stream/start on an active stream: adopt the new duck/volume/override_mute + // params and re-apply the policy through on_announcement_start. The task keeps + // playing; no state change and no task signal. + ServerAnnouncementStreamObject stream_params; + if (this->event_state->stream_params_slot.take(stream_params)) { + this->current_stream_params = std::move(stream_params); + } + if (this->listener && this->stream_active) { + const uint32_t generation = this->cleanup_generation; + this->listener->on_announcement_start(this->current_stream_params); + if (this->cleanup_generation != generation) { + teardown_reentered = true; + } + } + break; + } + case AnnouncementStreamCallbackType::OUTPUT_STARTED: { + if (!this->announcement_playing) { + this->announcement_playing = true; + state_changed = true; + } + break; + } + case AnnouncementStreamCallbackType::OUTPUT_FINISHED: { + // The task finished playing out the announcement: a graceful server stream/end + // that drained the buffered tail, or the stall guard firing without one. This is + // where the ducking is released for those paths (a transport-loss STREAM_END has + // already released it, so the stream_active guard makes this a no-op there). + if (this->listener && this->stream_active) { + const uint32_t generation = this->cleanup_generation; + this->listener->on_announcement_end(); + if (this->cleanup_generation != generation) { + this->stream_active = false; + teardown_reentered = true; + break; + } + } + this->stream_active = false; + if (this->announcement_playing) { + this->announcement_playing = false; + state_changed = true; + } + break; + } + } + if (teardown_reentered) { + break; + } + ++processed; + } + + // Clamp: 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)); + } + + if (state_changed) { + this->client->publish_state(); + } +} + +void AnnouncementRole::Impl::cleanup() { + // Flag the teardown for a drain_events() frame that may be on the call stack right now. + this->cleanup_generation++; + + // End any in-flight announcement: the task drains and returns to idle. + this->task->signal_stream_end(); + + // Discard stale slot content from the dead connection. Stale ring-borne events were already + // discarded by SendspinClient::cleanup_connection_state()'s inbox.reset_events() call. + this->event_state->stream_params_slot.reset(); + + // Clear pending events (main-thread only), then enqueue a clean STREAM_END so drain_events() + // releases the ducking and reports idle exactly once. + this->pending_events.clear(); + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_END); +} + +// ============================================================================ +// Impl: Helpers +// ============================================================================ + +void AnnouncementRole::Impl::enqueue_stream_event(AnnouncementStreamCallbackType event) const { + // A dropped STREAM_START would leave the task waiting for its start signal; a dropped + // STREAM_END would leave media ducked. Both wedge the announcement, so log drops at ERROR. + static const char* const EVENT_NAMES[] = {"STREAM_START", "STREAM_END", "CONFIG_UPDATE", + "OUTPUT_STARTED", "OUTPUT_FINISHED"}; + push_event_or_log(this->inbox, InboxEventType::ANNOUNCEMENT_STREAM, static_cast(event), + TAG, EVENT_NAMES[static_cast(event)], + /*error_level=*/true); +} + +} // namespace sendspin diff --git a/src/announcement_role_impl.h b/src/announcement_role_impl.h new file mode 100644 index 00000000..ba047ae4 --- /dev/null +++ b/src/announcement_role_impl.h @@ -0,0 +1,116 @@ +// 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 announcement_role_impl.h +/// @brief Private implementation for the announcement role (pimpl) + +#pragma once + +#include "announcement_task.h" +#include "inbox.h" +#include "sendspin/announcement_role.h" + +#include +#include + +namespace sendspin { + +class SendspinClient; +struct ClientHelloMessage; +struct ClientStateMessage; + +/// @brief Deferred announcement lifecycle callback types queued from the network and task threads +enum class AnnouncementStreamCallbackType : uint8_t { + STREAM_START, // New announcement stream is starting (from the network thread) + STREAM_END, // Announcement stream discarded and ended (from cleanup / a failed start) + CONFIG_UPDATE, // Re-sent stream/start updates the active stream's config (network thread) + OUTPUT_STARTED, // Task began writing announcement audio to the sink (from the task thread) + OUTPUT_FINISHED, // Task finished playing out the announcement (from the task thread) +}; + +/// @brief Private implementation of the announcement role +struct AnnouncementRole::Impl { + Impl(AnnouncementRoleConfig config, SendspinClient* client); + ~Impl(); + + // ======================================== + // Event state + // ======================================== + + struct EventState { + InboxSlot stream_params_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_binary(const uint8_t* data, size_t len) const; + void handle_stream_start(const ServerAnnouncementStreamObject& announcement_obj) const; + void handle_stream_end() const; + void on_stream_ring_event(AnnouncementStreamCallbackType event); + // True if this tick has drainable announcement work. All announcement events (lifecycle from + // the network thread, output transitions from the announcement task) travel over the shared + // event ring and are appended to pending_events by on_stream_ring_event() during this tick's + // ring dispatch, so the vector is the single source of pending work. stream_params_slot needs + // no separate topic-bit term: it is only consumed from the STREAM_START branch, which the + // pending_events term covers. + bool needs_drain(uint32_t /*pending_bits*/) const { + return !this->pending_events.empty(); + } + void drain_events(); + void cleanup(); + + // ======================================== + // Helpers + // ======================================== + + void enqueue_stream_event(AnnouncementStreamCallbackType event) const; + + // ======================================== + // Fields + // ======================================== + + // Struct fields + AnnouncementRoleConfig config; + ServerAnnouncementStreamObject current_stream_params{}; + std::vector pending_events; + + // Pointer fields + SendspinClient* client; + std::unique_ptr event_state; + Inbox* inbox{nullptr}; + AnnouncementRoleListener* listener{nullptr}; + std::unique_ptr task; + + // 32-bit fields + // Bumped by cleanup() so a drain_events() listener callback that re-enters connection + // teardown is detected when control returns (mirrors the player role). Main-thread only. + uint32_t cleanup_generation{0}; + + // 8-bit fields + // True while the announcement pipeline is outputting audio; reported to the server via + // client/state.announcement. Main-thread only. + bool announcement_playing{false}; + // True between the drained STREAM_START and the stream's end (server end, local completion, + // or cleanup); keeps on_announcement_end() paired 1:1 with on_announcement_start(). + // Main-thread only. + bool stream_active{false}; +}; + +} // namespace sendspin diff --git a/src/announcement_task.cpp b/src/announcement_task.cpp new file mode 100644 index 00000000..832fcb07 --- /dev/null +++ b/src/announcement_task.cpp @@ -0,0 +1,416 @@ +// 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 "announcement_task.h" + +#include "announcement_role_impl.h" +#include "platform/logging.h" +#include "platform/thread.h" +#include "platform/time.h" +#include "sendspin/client.h" + +#include + +namespace sendspin { + +static const char* const TAG = "sendspin.announcement_task"; + +/// @brief Task stack size; Opus decode dominates, same budget as the sync task +static constexpr size_t ANNOUNCEMENT_TASK_STACK_SIZE = 6192; + +/// @brief Timeout (ms) for receiving the next encoded chunk from the ring buffer +static constexpr uint32_t ANNOUNCEMENT_CHUNK_RECEIVE_TIMEOUT_MS = 15U; + +/// @brief Wait slice (ms) while waiting for the main loop's stream start acknowledgement +static constexpr uint32_t START_ACK_WAIT_MS = 20U; + +/// @brief Timeout (ms) for on_announcement_write pushes; bounds how long the task blocks on the +/// sink before re-checking command flags +static constexpr uint32_t ANNOUNCEMENT_WRITE_TIMEOUT_MS = 20U; + +/// @brief Wait slice (ms) while sleeping toward the first chunk's start time +static constexpr uint32_t START_WAIT_SLICE_MS = 10U; + +/// @brief Underrun guard: an open announcement stream whose buffer has drained with no data +/// arriving for this long is ended locally (5 s, per the spec's recommendation) +static constexpr int64_t ANNOUNCEMENT_STALL_TIMEOUT_US = 5000000; + +/// @brief Upper bound on how long the task waits for the first chunk's timestamp; a farther +/// future timestamp starts playback anyway (announcements are near-now by design) +static constexpr int64_t ANNOUNCEMENT_MAX_START_WAIT_US = 5000000; + +// ============================================================================ +// Lifecycle +// ============================================================================ + +AnnouncementTask::~AnnouncementTask() { + this->stop(); +} + +bool AnnouncementTask::init(AnnouncementRole::Impl* announcement_impl, SendspinClient* client, + size_t buffer_size) { + this->announcement_impl_ = announcement_impl; + this->client_ = client; + + if (!this->event_flags_.create()) { + SS_LOGE(TAG, "Couldn't create event flags."); + return false; + } + + this->encoded_ring_buffer_ = SendspinAudioRingBuffer::create(buffer_size); + if (this->encoded_ring_buffer_ == nullptr) { + SS_LOGE(TAG, "Couldn't create encoded announcement ring buffer."); + return false; + } + + this->decoder_ = std::make_unique(); + + return true; +} + +bool AnnouncementTask::start(bool task_stack_in_psram, unsigned priority) { + if (!this->is_initialized()) { + SS_LOGE(TAG, "Announcement task not initialized (call init() first)"); + return false; + } + + if (this->task_thread_.joinable()) { + SS_LOGW(TAG, "Announcement task thread already started"); + return false; + } + + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_TASK_RUNNING | + AnnouncementTaskBits::ANNOUNCEMENT_TASK_STOPPED | + AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START); + + platform_configure_thread("SendspinAnnc", ANNOUNCEMENT_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(AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE | + AnnouncementTaskBits::ANNOUNCEMENT_TASK_STOPPED, + false, false, UINT32_MAX); + + return true; +} + +// ============================================================================ +// Public API +// ============================================================================ + +void AnnouncementTask::signal_stream_end() { + // Lifecycle signals arrive unconditionally from the role, but init() only runs when the + // role has announcement formats and a listener; setting bits on uncreated event flags is a + // null-handle crash on ESP, so all signal/query paths guard on is_initialized(). + if (!this->is_initialized()) { + return; + } + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END); +} + +void AnnouncementTask::signal_stream_drain() { + if (!this->is_initialized()) { + return; + } + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN); +} + +void AnnouncementTask::signal_stream_start() { + if (!this->is_initialized()) { + return; + } + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START); +} + +bool AnnouncementTask::write_audio_chunk(const uint8_t* data, size_t data_size, int64_t timestamp, + ChunkType chunk_type, uint32_t timeout_ms) { + if (this->encoded_ring_buffer_ == nullptr) { + return false; + } + return this->encoded_ring_buffer_->write_chunk(data, data_size, timestamp, chunk_type, + timeout_ms); +} + +// ============================================================================ +// Task loop +// ============================================================================ + +void AnnouncementTask::thread_entry(void* params) { + auto* task = static_cast(params); + task->run(); +} + +void AnnouncementTask::run() { + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE); + + while ((this->event_flags_.get() & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP) == 0U) { + AudioStreamInfo stream_info{}; + if (!this->wait_for_codec_header(&stream_info)) { + break; // COMMAND_STOP + } + + // Wait for the main loop's start acknowledgement so on_announcement_start() (where the + // embedder applies the ducking) fires before any announcement audio reaches the sink. + bool start_acked = false; + while (true) { + const uint32_t flags = this->event_flags_.get(); + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP) != 0U) { + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END) != 0U) { + // Discarded before playback started (transport loss): drop the buffer and abort. + // A pending drain is superseded by the discard. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN); + this->discard_ring_buffer(); + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START) != 0U) { + // A drain that raced ahead of the start ack stays set and is honored by + // play_stream(), which plays the buffered audio out before ending. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START); + start_acked = true; + break; + } + this->event_flags_.wait(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END, + false, false, START_ACK_WAIT_MS); + } + if (!start_acked) { + this->decoder_->reset_decoders(); + continue; + } + + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE); + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_TASK_RUNNING); + + this->play_stream(); + + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_TASK_RUNNING); + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE); + this->decoder_->reset_decoders(); + this->announcement_impl_->enqueue_stream_event( + AnnouncementStreamCallbackType::OUTPUT_FINISHED); + } + + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE | + AnnouncementTaskBits::ANNOUNCEMENT_TASK_RUNNING); + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_TASK_STOPPED); +} + +bool AnnouncementTask::wait_for_codec_header(AudioStreamInfo* stream_info) { + while (true) { + const uint32_t flags = this->event_flags_.get(); + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP) != 0U) { + return false; + } + if ((flags & (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN)) != 0U) { + // Stale end/drain command from a stream that ended while idle; clearing it here keeps + // it from leaking into the next stream once a codec header arrives. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN); + } + + AudioRingBufferEntry* entry = + this->encoded_ring_buffer_->receive_chunk(ANNOUNCEMENT_CHUNK_RECEIVE_TIMEOUT_MS); + if (entry == nullptr) { + continue; + } + + const ChunkType chunk_type = entry->chunk_type; + if (chunk_type == CHUNK_TYPE_ENCODED_AUDIO) { + // Stale audio from a previous announcement; discard until a codec header arrives + this->encoded_ring_buffer_->return_chunk(entry); + continue; + } + + // The header chunk carries the stream's single scheduled start time. + const int64_t start_timestamp = entry->timestamp; + const bool header_ok = this->decoder_->process_header(entry->data(), entry->data_size, + chunk_type, stream_info); + this->encoded_ring_buffer_->return_chunk(entry); + if (header_ok) { + this->start_timestamp_ = start_timestamp; + this->decode_buffer_.resize(this->decoder_->get_decode_buffer_size()); + return true; + } + SS_LOGE(TAG, "Failed to process announcement codec header"); + } +} + +void AnnouncementTask::play_stream() { + bool output_started = false; + bool draining = false; + int64_t last_data_us = platform_time_us(); + + while (true) { + const uint32_t flags = this->event_flags_.get(); + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP) != 0U) { + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END) != 0U) { + // Non-graceful end (transport loss): discard any remaining buffered audio and stop. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END); + this->discard_ring_buffer(); + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN) != 0U) { + // Graceful server stream/end: stop expecting new audio and play out what is buffered. + // The loop keeps decoding and writing until the ring buffer empties (below), then ends. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_DRAIN); + draining = true; + } + + AudioRingBufferEntry* entry = + this->encoded_ring_buffer_->receive_chunk(ANNOUNCEMENT_CHUNK_RECEIVE_TIMEOUT_MS); + if (entry == nullptr) { + if (draining) { + break; // Graceful end: the buffered tail has played out + } + if (platform_time_us() - last_data_us > ANNOUNCEMENT_STALL_TIMEOUT_US) { + SS_LOGW(TAG, "Announcement stream stalled; ending locally"); + break; + } + continue; + } + last_data_us = platform_time_us(); + + if (entry->chunk_type != CHUNK_TYPE_ENCODED_AUDIO) { + // Mid-stream codec header (configuration update): re-initialize the decoder + AudioStreamInfo new_info{}; + if (this->decoder_->process_header(entry->data(), entry->data_size, entry->chunk_type, + &new_info)) { + this->decode_buffer_.resize(this->decoder_->get_decode_buffer_size()); + } else { + SS_LOGE(TAG, "Failed to process mid-stream announcement codec header"); + } + this->encoded_ring_buffer_->return_chunk(entry); + continue; + } + + size_t decoded_size = 0; + bool decode_ok = this->decoder_->decode_audio_chunk( + entry->data(), entry->data_size, this->decode_buffer_.data(), + this->decode_buffer_.size(), &decoded_size); + if (!decode_ok && this->decoder_->get_decode_buffer_size() > this->decode_buffer_.size()) { + // Opus decode-buffer growth: enlarge and retry once + this->decode_buffer_.resize(this->decoder_->get_decode_buffer_size()); + decode_ok = this->decoder_->decode_audio_chunk( + entry->data(), entry->data_size, this->decode_buffer_.data(), + this->decode_buffer_.size(), &decoded_size); + } + this->encoded_ring_buffer_->return_chunk(entry); + + if (!decode_ok) { + SS_LOGW(TAG, "Failed to decode announcement chunk"); + continue; + } + if (decoded_size == 0) { + continue; + } + + if (!output_started) { + // Loose start scheduling: begin at the stream's single start_timestamp when time sync + // is available, or as soon as possible otherwise. Announcement chunks are never + // dropped for lateness. + this->wait_for_start_time(this->start_timestamp_); + if ((this->event_flags_.get() & + (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END)) != 0U) { + continue; // Re-enter the loop head to handle the command + } + output_started = true; + this->announcement_impl_->enqueue_stream_event( + AnnouncementStreamCallbackType::OUTPUT_STARTED); + } + + // Push the decoded PCM to the sink; the blocking write is what paces this loop + AnnouncementRoleListener* listener = this->announcement_impl_->listener; + size_t offset = 0; + while (offset < decoded_size && listener != nullptr) { + if ((this->event_flags_.get() & + (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END)) != 0U) { + break; + } + const size_t written = listener->on_announcement_write( + this->decode_buffer_.data() + offset, decoded_size - offset, + ANNOUNCEMENT_WRITE_TIMEOUT_MS); + if (written == 0) { + if (platform_time_us() - last_data_us > ANNOUNCEMENT_STALL_TIMEOUT_US) { + SS_LOGW(TAG, "Announcement sink stalled; ending locally"); + return; + } + continue; + } + last_data_us = platform_time_us(); + offset += written; + } + } +} + +void AnnouncementTask::wait_for_start_time(int64_t server_timestamp) { + if (this->client_ == nullptr || !this->client_->is_time_synced()) { + return; + } + const int64_t target_us = this->client_->get_client_time(server_timestamp); + const int64_t wait_deadline_us = platform_time_us() + ANNOUNCEMENT_MAX_START_WAIT_US; + + while (true) { + if ((this->event_flags_.get() & (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END)) != + 0U) { + return; + } + const int64_t now_us = platform_time_us(); + if (now_us >= target_us || now_us >= wait_deadline_us) { + return; + } + const int64_t remaining_ms = (std::min(target_us, wait_deadline_us) - now_us) / 1000; + const uint32_t slice_ms = + static_cast(std::min(remaining_ms + 1, START_WAIT_SLICE_MS)); + this->event_flags_.wait(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END, + false, false, slice_ms); + } +} + +void AnnouncementTask::discard_ring_buffer() { + while (true) { + AudioRingBufferEntry* entry = this->encoded_ring_buffer_->receive_chunk(0); + if (entry == nullptr) { + return; + } + this->encoded_ring_buffer_->return_chunk(entry); + } +} + +void AnnouncementTask::stop() { + if (this->is_initialized()) { + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP); + } + if (this->task_thread_.joinable()) { + this->task_thread_.join(); + } +} + +} // namespace sendspin diff --git a/src/announcement_task.h b/src/announcement_task.h new file mode 100644 index 00000000..1bb8fbe7 --- /dev/null +++ b/src/announcement_task.h @@ -0,0 +1,172 @@ +// 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 announcement_task.h +/// @brief Background task that decodes announcement audio and writes PCM to the announcement sink + +#pragma once + +#include "audio_ring_buffer.h" +#include "audio_stream_info.h" +#include "decoder.h" +#include "platform/event_flags.h" +#include "sendspin/announcement_role.h" + +#include +#include +#include + +namespace sendspin { + +class SendspinClient; + +/// @brief Event flag bits used for announcement task lifecycle and command signaling +/// +/// Kept distinct from the sync task's EventGroupBits: both enums are unscoped in this namespace +/// and may be visible in the same translation unit. +enum AnnouncementTaskBits : uint16_t { + ANNOUNCEMENT_COMMAND_STOP = (1 << 0), // Signal task to stop + ANNOUNCEMENT_COMMAND_STREAM_END = (1 << 1), // Discard the stream and end it (transport loss) + ANNOUNCEMENT_COMMAND_STREAM_DRAIN = (1 << 2), // Play out buffered audio, then end the stream + ANNOUNCEMENT_COMMAND_START = (1 << 3), // Signal stream start acknowledged + ANNOUNCEMENT_TASK_RUNNING = (1 << 8), // Task is actively playing an announcement + ANNOUNCEMENT_TASK_STOPPED = (1 << 10), // Task thread has exited + ANNOUNCEMENT_TASK_IDLE = (1 << 12), // Task is idle, waiting for an announcement +}; + +/// @brief Self-contained background task for announcement playback +/// +/// Manages a persistent background thread that reads encoded announcement audio from a +/// dedicated ring buffer, decodes it, and writes PCM data via the announcement listener. +/// Unlike the player's SyncTask there is no sample-accurate sync machinery: announcements are +/// per-client, output begins at (or as soon as possible after) the stream's single +/// start_timestamp, and subsequent output is paced by the sink's blocking writes. A stalled +/// stream (buffer drained, no data arriving) is ended locally after a timeout. +/// +/// The thread starts once during initialization and idles between announcements to avoid +/// thread create/destroy churn on embedded devices. +class AnnouncementTask { +public: + AnnouncementTask() = default; + ~AnnouncementTask(); + + /// @brief Initializes event flags and creates the encoded ring buffer + /// @param announcement_impl The owning AnnouncementRole::Impl, used for listener and events. + /// @param client The owning SendspinClient, used for time conversion. + /// @param buffer_size Size of the encoded announcement ring buffer in bytes. + /// @return true on success, false on allocation failure. + bool init(AnnouncementRole::Impl* announcement_impl, SendspinClient* client, + size_t buffer_size); + + /// @brief Creates and starts the persistent announcement background thread + /// Call once after init(). The thread idles until a codec header arrives in the ring buffer. + /// @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 announcement task has been initialized, false otherwise. + bool is_initialized() const { + return this->event_flags_.is_created(); + } + + /// @brief Returns true if the announcement task is actively playing an announcement + /// @return true if actively decoding and playing, false when idle or stopped. + bool is_running() const { + // Guarded so callers may query before init(); reading uncreated event flags is a + // null-handle crash on ESP + if (!this->is_initialized()) { + return false; + } + return (this->event_flags_.get() & AnnouncementTaskBits::ANNOUNCEMENT_TASK_RUNNING) != 0U; + } + + /// @brief Signals the task to discard the current announcement stream and end it. Non-blocking + /// The task drops any audio still in the ring buffer and returns to idle. Used for + /// non-graceful teardown (transport loss), where the buffered tail cannot or should not play. + /// Thread-safe: may be called from any context. + void signal_stream_end(); + + /// @brief Signals the task to stop expecting new audio and play out what remains. Non-blocking + /// The task keeps decoding and writing the buffered audio until the ring buffer is empty, then + /// ends the stream (which releases the ducking via OUTPUT_FINISHED). Used for a graceful + /// server stream/end so the announcement tail is not cut off. + /// Thread-safe: may be called from any context. + void signal_stream_drain(); + + /// @brief Signals the task that the client has processed the announcement stream start + /// The task waits for this after finding a codec header, so the main-thread + /// on_announcement_start() callback (where the embedder applies ducking) fires before any + /// announcement audio is written to the sink. + /// Thread-safe: may be called from any context. + void signal_stream_start(); + + /// @brief Writes an encoded announcement chunk into the ring buffer + /// Called from the client's binary message callback (may be any thread). + /// @param data Pointer to the audio data. + /// @param data_size Size of the audio data in bytes. + /// @param timestamp The stream's start_timestamp when writing the codec header (the task + /// reads the scheduled start time from it); 0 for the untimed audio chunks. + /// @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. + bool write_audio_chunk(const uint8_t* data, size_t data_size, int64_t timestamp, + ChunkType chunk_type, uint32_t timeout_ms); + +protected: + /// @brief Entry point for the persistent announcement background thread + /// @param params Pointer to the owning AnnouncementTask instance. + static void thread_entry(void* params); + + /// @brief Outer task loop: idle -> header -> start ack -> play -> idle + void run(); + + /// @brief Waits in IDLE for a codec header to arrive in the ring buffer + /// Discards stale audio chunks. Captures the stream's start_timestamp (carried on the header + /// chunk) into start_timestamp_. Returns true if a codec header was processed into the + /// decoder. Returns false if ANNOUNCEMENT_COMMAND_STOP was signaled. + bool wait_for_codec_header(AudioStreamInfo* stream_info); + + /// @brief Plays one announcement stream until end/stop/stall, writing PCM to the listener + void play_stream(); + + /// @brief Sleeps until the stream's start_timestamp (translated to the local clock), bounded + /// and abortable. No-op when time sync is unavailable or the timestamp already passed. + /// Honoring the timestamp is what aligns a coordinated multi-speaker announcement, where the + /// server schedules the same start time on every targeted client + void wait_for_start_time(int64_t server_timestamp); + + /// @brief Non-blocking discard of all chunks currently in the ring buffer + void discard_ring_buffer(); + + /// @brief Signals the task to stop and waits for the thread to finish + void stop(); + + // Struct fields + EventFlags event_flags_; + std::thread task_thread_; + std::vector decode_buffer_; + // The stream's single scheduled start time (server clock, microseconds), captured from the + // codec header chunk. Task-thread only. 0 means start as soon as possible. + int64_t start_timestamp_{0}; + + // Pointer fields + AnnouncementRole::Impl* announcement_impl_{nullptr}; + SendspinClient* client_{nullptr}; + std::unique_ptr decoder_; + std::unique_ptr encoded_ring_buffer_; +}; + +} // namespace sendspin diff --git a/src/client.cpp b/src/client.cpp index e5f77d07..bfa33cff 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -22,6 +22,9 @@ #include "platform/logging.h" #include "platform/memory.h" #include "platform/network_info.h" +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT +#include "announcement_role_impl.h" +#endif #ifdef SENDSPIN_ENABLE_ARTWORK #include "artwork_role_impl.h" #endif @@ -102,6 +105,9 @@ SendspinClient::~SendspinClient() { #ifdef SENDSPIN_ENABLE_PLAYER this->player_.reset(); #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + this->announcement_.reset(); +#endif #ifdef SENDSPIN_ENABLE_VISUALIZER this->visualizer_.reset(); #endif @@ -154,6 +160,11 @@ bool SendspinClient::start() { roles_started = this->player_->impl_->start(); } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (roles_started && this->announcement_) { + roles_started = this->announcement_->impl_->start(); + } +#endif #ifdef SENDSPIN_ENABLE_VISUALIZER if (roles_started && this->visualizer_) { roles_started = this->visualizer_->impl_->start(); @@ -424,6 +435,17 @@ void SendspinClient::drain_inbox() { this->visualizer_->impl_->handle_stream_ring_event( static_cast(event.code)); } +#endif + break; + } + // ANNOUNCEMENT_STREAM: lifecycle from the network thread plus output + // transitions from the announcement task, dispatched like PLAYER_STREAM. + case InboxEventType::ANNOUNCEMENT_STREAM: { +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + this->announcement_->impl_->on_stream_ring_event( + static_cast(event.code)); + } #endif break; } @@ -459,6 +481,11 @@ void SendspinClient::drain_inbox() { this->player_->impl_->drain_events(); } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_ && this->announcement_->impl_->needs_drain(slot_bits)) { + this->announcement_->impl_->drain_events(); + } +#endif #ifdef SENDSPIN_ENABLE_CONTROLLER if (this->controller_ && this->controller_->impl_->needs_drain(slot_bits)) { this->controller_->impl_->drain_events(); @@ -583,6 +610,17 @@ VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT +AnnouncementRole& SendspinClient::add_announcement(AnnouncementRoleConfig config) { + if (this->lifecycle_.load(std::memory_order_relaxed) != LifecycleState::STOPPED) { + SS_LOGW(TAG, "add_announcement() called while started; role may not initialize correctly"); + } + this->announcement_ = std::make_unique(std::move(config), this); + this->announcement_->impl_->attach_inbox(this->event_state_->inbox); + return *this->announcement_; +} +#endif + // ============================================================================ // Queries // ============================================================================ @@ -687,6 +725,11 @@ void SendspinClient::cleanup_connection_state() { this->player_->impl_->cleanup(); } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + this->announcement_->impl_->cleanup(); + } +#endif #ifdef SENDSPIN_ENABLE_CONTROLLER if (this->controller_) { this->controller_->impl_->cleanup(); @@ -777,6 +820,11 @@ std::string SendspinClient::build_hello_message() { this->visualizer_->impl_->build_hello_fields(msg); } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + this->announcement_->impl_->build_hello_fields(msg); + } +#endif return format_client_hello_message(&msg); } @@ -836,14 +884,25 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* this->visualizer_->impl_->handle_stream_start(stream_msg.visualizer.value()); } #endif + +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_ && stream_msg.announcement.has_value()) { + this->announcement_->impl_->handle_stream_start(stream_msg.announcement.value()); + } +#endif break; } case SendspinServerToClientMessageType::STREAM_END: { StreamEndMessage end_msg; if (process_stream_end_message(root, &end_msg)) { + // Omitted roles ends the media streams (player, artwork, visualizer). The + // announcement stream is independent: it ends only when the roles list explicitly + // names "announcement", mirroring how stream/clear excludes it. A bare stream/end + // therefore leaves an active announcement playing and its ducking intact. bool end_player = !end_msg.roles.has_value(); bool end_artwork = !end_msg.roles.has_value(); bool end_visualizer = !end_msg.roles.has_value(); + bool end_announcement = false; if (end_msg.roles.has_value()) { for (const auto& role : end_msg.roles.value()) { @@ -853,12 +912,14 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* end_artwork = true; } else if (role == "visualizer") { end_visualizer = true; + } else if (role == "announcement") { + end_announcement = true; } } } - SS_LOGD(TAG, "Stream ended - player:%d artwork:%d visualizer:%d", end_player, - end_artwork, end_visualizer); + SS_LOGD(TAG, "Stream ended - player:%d artwork:%d visualizer:%d announcement:%d", + end_player, end_artwork, end_visualizer, end_announcement); #ifdef SENDSPIN_ENABLE_PLAYER if (this->player_ && end_player) { @@ -877,12 +938,21 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* this->visualizer_->impl_->handle_stream_end(); } #endif + +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_ && end_announcement) { + this->announcement_->impl_->handle_stream_end(); + } +#endif } break; } case SendspinServerToClientMessageType::STREAM_CLEAR: { StreamClearMessage clear_msg; if (process_stream_clear_message(root, &clear_msg)) { + // stream/clear is media-scoped and never lists the announcement role: an active + // announcement stream is unaffected. Omitted roles is a media seek that clears the + // player and visualizer streams. bool clear_player = !clear_msg.roles.has_value(); bool clear_artwork = !clear_msg.roles.has_value(); bool clear_visualizer = !clear_msg.roles.has_value(); @@ -1064,6 +1134,18 @@ SS_HOT void SendspinClient::process_binary_message(const uint8_t* payload, size_ if (this->artwork_) { this->artwork_->impl_->handle_binary(slot, data, data_len); } +#endif + break; + } + case SENDSPIN_ROLE_ANNOUNCEMENT: { +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + if (slot == 0) { + this->announcement_->impl_->handle_binary(data, data_len); + } else { + SS_LOGW(TAG, "Unknown announcement binary slot %d", slot); + } + } #endif break; } @@ -1092,6 +1174,12 @@ void SendspinClient::publish_client_state(SendspinConnection* conn) { } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + this->announcement_->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/inbox.h b/src/inbox.h index 2f031f06..b3a882ec 100644 --- a/src/inbox.h +++ b/src/inbox.h @@ -48,6 +48,8 @@ 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_ANNOUNCEMENT_STREAM_PARAMS = + 1U << 10; // Announcement stream params slot // ============================================================================ // Event ring types @@ -58,13 +60,14 @@ static constexpr uint32_t INBOX_TOPIC_PLAYER_STATE = 1U << 9; // Player /// The `code` field on InboxEvent carries a role-local enum value (cast to/from uint8_t by the /// producer/consumer); the inbox does not interpret it. enum class InboxEventType : uint8_t { - TIME_RESPONSE, // Time-sync measurement; payload in InboxEvent::time - PLAYER_STREAM, // Player stream lifecycle; code = PlayerStreamCallbackType - CONTROLLER_CLEARED, // Controller state cleared on disconnect; no payload - METADATA_CLEARED, // Metadata cleared on disconnect; no payload - COLOR_CLEARED, // Color state cleared on disconnect; no payload - ARTWORK_STREAM, // Artwork stream lifecycle; code = ArtworkEventType - VISUALIZER_STREAM, // Visualizer stream lifecycle; code = VisualizerEventType + TIME_RESPONSE, // Time-sync measurement; payload in InboxEvent::time + PLAYER_STREAM, // Player stream lifecycle; code = PlayerStreamCallbackType + CONTROLLER_CLEARED, // Controller state cleared on disconnect; no payload + METADATA_CLEARED, // Metadata cleared on disconnect; no payload + COLOR_CLEARED, // Color state cleared on disconnect; no payload + ARTWORK_STREAM, // Artwork stream lifecycle; code = ArtworkEventType + VISUALIZER_STREAM, // Visualizer stream lifecycle; code = VisualizerEventType + ANNOUNCEMENT_STREAM, // Announcement stream lifecycle; code = AnnouncementStreamCallbackType }; /// @brief Payload for TIME_RESPONSE events diff --git a/src/protocol.cpp b/src/protocol.cpp index eace5644..d09f6c42 100644 --- a/src/protocol.cpp +++ b/src/protocol.cpp @@ -34,6 +34,12 @@ static const char* const TAG = "sendspin.protocol"; /// @brief Protocol maximum for volume fields (volume is 0-100 on the wire). static constexpr uint8_t VOLUME_MAX = 100; +/// @brief Protocol maximum for the announcement media_duck_db field (0-50 dB on the wire). +static constexpr uint8_t MEDIA_DUCK_DB_MAX = 50; + +/// @brief Protocol maximum for the announcement duck_ramp_ms field (0-2000 ms on the wire). +static constexpr uint16_t DUCK_RAMP_MS_MAX = 2000; + /// @brief Reads an optional unsigned-integer field with strict type and optional range validation. /// Absent or null returns nullopt silently (a legal state for an optional field). A present value /// that is the wrong JSON type, outside the target type's range, or outside [min, max] is logged @@ -690,6 +696,43 @@ bool process_stream_start_message(JsonObject root, StreamStartMessage* stream_ms } } + if (root["payload"]["announcement"].is()) { + JsonObject announcement_json = root["payload"]["announcement"]; + ServerAnnouncementStreamObject announcement_obj{}; + // The codec parameters share the player stream-object shape and validation + if (process_player_stream_object(announcement_json, &announcement_obj.format, true)) { + if (!announcement_obj.is_complete()) { + SS_LOGE(TAG, "Invalid stream/start message: incomplete announcement object"); + return false; + } + // start_timestamp is the single scheduled start time for the whole announcement and + // is required; reject the object if it is missing or not an integer. + if (!announcement_json["start_timestamp"].is()) { + SS_LOGE(TAG, "Invalid stream/start message: announcement missing start_timestamp"); + return false; + } + announcement_obj.start_timestamp = announcement_json["start_timestamp"].as(); + if (auto v = read_uint_field(announcement_json["media_duck_db"], + "media_duck_db", 0, MEDIA_DUCK_DB_MAX)) { + announcement_obj.media_duck_db = *v; + } + if (auto v = read_uint_field(announcement_json["duck_ramp_ms"], + "duck_ramp_ms", 0, DUCK_RAMP_MS_MAX)) { + announcement_obj.duck_ramp_ms = *v; + } + if (auto v = read_uint_field(announcement_json["volume"], "volume", 0, + VOLUME_MAX)) { + announcement_obj.volume = v; + } + if (auto v = read_bool_field(announcement_json["override_mute"], "override_mute")) { + announcement_obj.override_mute = *v; + } + stream_msg->announcement = std::move(announcement_obj); + } else { + return false; + } + } + return true; } @@ -883,6 +926,20 @@ std::string format_client_hello_message(const ClientHelloMessage* msg) { } } + if (msg->announcement_v1_support.has_value()) { + JsonArray formats_list = + root["payload"]["announcement@v1_support"]["supported_formats"].to(); + for (const auto& format : msg->announcement_v1_support.value().supported_formats) { + JsonObject format_obj = formats_list.add(); + format_obj["codec"] = to_cstr(format.codec); + format_obj["channels"] = format.channels; + format_obj["sample_rate"] = format.sample_rate; + format_obj["bit_depth"] = format.bit_depth; + } + root["payload"]["announcement@v1_support"]["buffer_capacity"] = + msg->announcement_v1_support.value().buffer_capacity; + } + std::string output; serializeJson(doc, output); return output; @@ -909,6 +966,24 @@ std::string format_client_state_message(const ClientStateMessage* msg) { } } + if (msg->announcement.has_value()) { + const ClientAnnouncementStateObject& announcement_state = msg->announcement.value(); + // Only create the "announcement" key when at least one field is set (all fields are + // optional on the wire) + if (announcement_state.playing.has_value() || + announcement_state.required_lead_time_ms.has_value()) { + JsonObject announcement_json = root["payload"]["announcement"].to(); + if (announcement_state.playing.has_value()) { + announcement_json["state"] = + announcement_state.playing.value() ? "playing" : "idle"; + } + if (announcement_state.required_lead_time_ms.has_value()) { + announcement_json["required_lead_time_ms"] = + announcement_state.required_lead_time_ms.value(); + } + } + } + std::string output; serializeJson(doc, output); return output; diff --git a/src/protocol_messages.h b/src/protocol_messages.h index 8fdeedb2..5784227c 100644 --- a/src/protocol_messages.h +++ b/src/protocol_messages.h @@ -18,6 +18,7 @@ #pragma once +#include "sendspin/announcement_role.h" #include "sendspin/artwork_role.h" #include "sendspin/color_role.h" #include "sendspin/controller_role.h" @@ -45,26 +46,28 @@ namespace sendspin { /// The visualizer role has an expanded 8-slot allocation (IDs 16-23, bits 2-0 as slot) /// and is dispatched by ID range rather than through this enum. enum SendspinBinaryRole : uint8_t { - SENDSPIN_ROLE_PLAYER = 1, // 000001xx (IDs 4-7) - SENDSPIN_ROLE_ARTWORK = 2, // 000010xx (IDs 8-11) + SENDSPIN_ROLE_PLAYER = 1, // 000001xx (IDs 4-7) + SENDSPIN_ROLE_ARTWORK = 2, // 000010xx (IDs 8-11) + SENDSPIN_ROLE_ANNOUNCEMENT = 6, // 000110xx (IDs 24-27) }; /// @brief Extracts the role field from a standard 4-slot binary message type byte /// @param type Binary message type byte. /// @return Role portion of the type (bits 7-2). -/// @warning Valid only for the standard 4-slot roles (PLAYER/ARTWORK, IDs 4-11). The visualizer -/// range (IDs 16-23) is dispatched by range in SendspinClient::process_binary_message -/// and must not be routed through this helper: get_binary_role(16) yields 4, which -/// matches no SendspinBinaryRole enumerator. +/// @warning Valid only for the standard 4-slot roles (PLAYER/ARTWORK IDs 4-11, ANNOUNCEMENT IDs +/// 24-27). The visualizer range (IDs 16-23) is dispatched by range in +/// SendspinClient::process_binary_message and must not be routed through this helper: +/// get_binary_role(16) yields 4, which matches no SendspinBinaryRole enumerator. inline uint8_t get_binary_role(uint8_t type) { return type >> 2; } /// @brief Extracts the slot field from a standard 4-slot binary message type byte /// @param type Binary message type byte. /// @return Slot portion of the type (bits 1-0). -/// @warning Valid only for the standard 4-slot roles (PLAYER/ARTWORK, IDs 4-11). It masks bits -/// 1-0, so it cannot address the visualizer's 8-slot range (e.g. IDs 16 and 20 both -/// alias to slot 0); those messages are dispatched by range, not by slot. +/// @warning Valid only for the standard 4-slot roles (PLAYER/ARTWORK IDs 4-11, ANNOUNCEMENT IDs +/// 24-27). It masks bits 1-0, so it cannot address the visualizer's 8-slot range (e.g. +/// IDs 16 and 20 both alias to slot 0); those messages are dispatched by range, not by +/// slot. inline uint8_t get_binary_slot(uint8_t type) { return type & 0x03; } @@ -82,6 +85,7 @@ enum SendspinBinaryType : uint8_t { SENDSPIN_BINARY_VISUALIZER_PEAK = 20, // uint8 onset strength SENDSPIN_BINARY_VISUALIZER_FIRST = 16, // Start of visualizer ID range SENDSPIN_BINARY_VISUALIZER_LAST = 23, // End of visualizer ID range (21-23 reserved) + SENDSPIN_BINARY_ANNOUNCEMENT_AUDIO = 24, // Announcement slot 0: encoded audio chunk }; /// @brief JSON message types sent from the server to the client @@ -99,12 +103,13 @@ enum class SendspinServerToClientMessageType : uint8_t { /// @brief Protocol role identifiers used in hello messages and role negotiation enum class SendspinRole : uint8_t { - PLAYER, // Audio playback role - CONTROLLER, // Playback command/state role - METADATA, // Track metadata role - ARTWORK, // Album artwork role - VISUALIZER, // Audio visualization role - COLOR, // Audio-derived color palette role + PLAYER, // Audio playback role + CONTROLLER, // Playback command/state role + METADATA, // Track metadata role + ARTWORK, // Album artwork role + VISUALIZER, // Audio visualization role + COLOR, // Audio-derived color palette role + ANNOUNCEMENT, // Per-client announcement audio role }; /// @brief Converts a SendspinRole value to its protocol wire string representation @@ -124,6 +129,8 @@ inline const char* to_cstr(SendspinRole role) { return "visualizer@v1"; case SendspinRole::COLOR: return "color@v1"; + case SendspinRole::ANNOUNCEMENT: + return "announcement@v1"; default: return "unknown"; } @@ -294,6 +301,23 @@ struct ClientPlayerStateObject { std::vector supported_commands{}; }; +// --- announcement_role.h --- + +/// @brief Announcement capabilities advertised to the server during the hello handshake +struct AnnouncementSupportObject { + std::vector supported_formats{}; + size_t buffer_capacity{}; +}; + +/// @brief Announcement state reported by the client to the server in client/state messages +/// +/// All fields are optional on the wire; the server derives announcement completion from the +/// stream timeline regardless, so this state is informational. +struct ClientAnnouncementStateObject { + std::optional playing{}; + std::optional required_lead_time_ms{}; +}; + // --- controller_role.h --- inline const char* to_cstr(SendspinControllerCommand cmd) { @@ -619,12 +643,14 @@ struct ClientHelloMessage { std::optional player_v1_support{}; std::optional artwork_v1_support{}; std::optional visualizer_support{}; + std::optional announcement_v1_support{}; }; /// @brief Outgoing client/state message reporting client playback state to the server struct ClientStateMessage { SendspinClientState state{}; std::optional player{}; + std::optional announcement{}; }; /// @brief Parsed server/hello handshake message received at connection startup @@ -645,6 +671,7 @@ struct StreamStartMessage { std::optional player; std::optional artwork; std::optional visualizer; + std::optional announcement; }; /// @brief Outgoing stream/request_format message used for codec, artwork, and visualizer diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index cd67a861..c2e87015 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -34,6 +34,7 @@ add_executable(sendspin_tests test_artwork_role.cpp test_client_teardown.cpp test_client_lifecycle.cpp + test_announcement_role.cpp ) # test_client_lifecycle.cpp checks what a role's stop() left in its ring, which nothing on the diff --git a/tests/test_announcement_role.cpp b/tests/test_announcement_role.cpp new file mode 100644 index 00000000..d64cf64e --- /dev/null +++ b/tests/test_announcement_role.cpp @@ -0,0 +1,553 @@ +// 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. + +// Unit tests for the announcement role's wire-protocol surface: the announcement@v1 hello +// support object, the stream/start announcement object with its ducking/volume fields, the +// client/state announcement object, and the binary type allocation. + +#include "announcement_role_impl.h" +#include "inbox.h" +#include "protocol_messages.h" +#include "sendspin/client.h" +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace sendspin; // NOLINT(google-build-using-namespace) -- test-local convenience + +namespace { + +// Parses a JSON string and returns the root object via the out-parameter, keeping the backing +// document alive in the caller. Returns false if the JSON is malformed. +bool parse(const std::string& json, JsonDocument& doc, JsonObject& root) { + if (deserializeJson(doc, json)) { + return false; + } + root = doc.as(); + return true; +} + +// Records the announcement lifecycle callbacks the role delivers to the embedder. +struct RecordingListener : AnnouncementRoleListener { + size_t on_announcement_write(uint8_t* /*data*/, size_t /*length*/, + uint32_t /*timeout_ms*/) override { + return 0; + } + void on_announcement_start(const ServerAnnouncementStreamObject& params) override { + ++start_count; + last_params = params; + } + void on_announcement_end() override { + ++end_count; + } + + int start_count{0}; + int end_count{0}; + ServerAnnouncementStreamObject last_params{}; +}; + +// Builds an announcement role Impl bound to a program-lifetime Inbox and a real, never-started +// SendspinClient. A default-constructed client never opens a connection, so publish_state() is a +// no-op (no current connection) and the task's time conversion reports "not synced" -- both of +// which these tests rely on. Client and Inbox are non-movable/have program lifetime via static +// deques (which never relocate existing elements), giving stable addresses that outlive the +// returned Impl, which only holds pointers to them. +std::unique_ptr make_announcement_impl() { + static std::deque clients; + static std::deque inboxes; + + AnnouncementRoleConfig config; + config.audio_formats.push_back({SendspinCodecFormat::PCM, 1, 48000, 16}); + clients.emplace_back(SendspinClientConfig{}); + auto impl = std::make_unique(std::move(config), &clients.back()); + inboxes.emplace_back(); + impl->attach_inbox(inboxes.back()); + return impl; +} + +constexpr auto POSITIVE_TIMEOUT = std::chrono::milliseconds(2000); + +// A PCM stream/start parameter object; the task builds a synthetic PCM header from it. PCM decode +// is a passthrough memcpy, so bytes fed via handle_binary() reach the listener unchanged. +ServerAnnouncementStreamObject make_pcm_stream_params() { + ServerAnnouncementStreamObject params{}; + params.format.codec = SendspinCodecFormat::PCM; + params.format.sample_rate = 48000; + params.format.channels = 1; + params.format.bit_depth = 16; + params.start_timestamp = 0; + return params; +} + +// A 128-byte PCM chunk (64 frames of mono s16) filled with a per-chunk marker byte, so the +// concatenation the listener records can be compared byte-for-byte against what was fed. +std::vector make_pcm_chunk(uint8_t marker) { + return std::vector(128, marker); +} + +// Records the PCM the role writes to the sink and every lifecycle callback. on_announcement_write +// runs on the task thread; the drain/end callbacks run on the "main loop" (test) thread via +// pump_announcement(). A gate lets a test hold the task inside its first write so later chunks +// stay buffered in the ring when a stream/end arrives. +struct DrainRecordingListener : AnnouncementRoleListener { + size_t on_announcement_write(uint8_t* data, size_t length, uint32_t /*timeout_ms*/) override { + std::unique_lock lock(this->mutex); + if (this->gate_enabled && !this->gate_released) { + this->first_write_reached = true; + this->cv.notify_all(); + this->cv.wait(lock, [&] { return this->gate_released; }); + } + this->written.insert(this->written.end(), data, data + length); + this->cv.notify_all(); + return length; + } + void on_announcement_start(const ServerAnnouncementStreamObject& /*params*/) override { + std::lock_guard lock(this->mutex); + ++this->start_count; + } + void on_announcement_end() override { + std::lock_guard lock(this->mutex); + ++this->end_count; + this->written_bytes_at_end = this->written.size(); + } + + bool wait_first_write(std::chrono::milliseconds timeout) { + std::unique_lock lock(this->mutex); + return this->cv.wait_for(lock, timeout, [&] { return this->first_write_reached; }); + } + void release_gate() { + { + std::lock_guard lock(this->mutex); + this->gate_released = true; + } + this->cv.notify_all(); + } + std::vector written_snapshot() { + std::lock_guard lock(this->mutex); + return this->written; + } + int starts() { + std::lock_guard lock(this->mutex); + return this->start_count; + } + int ends() { + std::lock_guard lock(this->mutex); + return this->end_count; + } + size_t end_bytes() { + std::lock_guard lock(this->mutex); + return this->written_bytes_at_end; + } + + std::mutex mutex; + std::condition_variable cv; + std::vector written; + bool gate_enabled{false}; + bool gate_released{false}; + bool first_write_reached{false}; + int start_count{0}; + int end_count{0}; + size_t written_bytes_at_end{0}; +}; + +// Mimics one SendspinClient::loop() tick for the announcement role: drains the shared event ring +// (the only path the announcement task's OUTPUT_STARTED/OUTPUT_FINISHED and the network-thread +// lifecycle events travel) into the role, then runs the main-thread event drain. +void pump_announcement(AnnouncementRole::Impl& impl) { + if ((impl.inbox->poll() & INBOX_TOPIC_EVENTS) != 0U) { + InboxEvent events[Inbox::EVENT_CAPACITY]; + const size_t count = impl.inbox->take_events(events, Inbox::EVENT_CAPACITY); + for (size_t i = 0; i < count; ++i) { + if (events[i].type == InboxEventType::ANNOUNCEMENT_STREAM) { + impl.on_stream_ring_event( + static_cast(events[i].code)); + } + } + } + impl.drain_events(); +} + +// Pumps the announcement role until `pred` holds or the timeout elapses. +template +bool pump_until(AnnouncementRole::Impl& impl, Pred pred, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + do { + pump_announcement(impl); + if (pred()) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } while (std::chrono::steady_clock::now() < deadline); + return pred(); +} + +} // namespace + +// ============================================================================ +// Role identity and binary allocation +// ============================================================================ + +TEST(AnnouncementRole, RoleWireString) { + EXPECT_STREQ(to_cstr(SendspinRole::ANNOUNCEMENT), "announcement@v1"); +} + +// The announcement role occupies binary role 6 (type IDs 24-27), the block after the +// visualizer's expanded 16-23 range; the source role holds 12-15. +TEST(AnnouncementRole, BinaryTypeAllocation) { + EXPECT_EQ(SENDSPIN_BINARY_ANNOUNCEMENT_AUDIO, 24); + EXPECT_EQ(get_binary_role(SENDSPIN_BINARY_ANNOUNCEMENT_AUDIO), SENDSPIN_ROLE_ANNOUNCEMENT); + EXPECT_EQ(get_binary_slot(SENDSPIN_BINARY_ANNOUNCEMENT_AUDIO), 0); +} + +// ============================================================================ +// client/hello announcement@v1_support serialization +// ============================================================================ + +TEST(AnnouncementRole, HelloSupportObjectSerialization) { + ClientHelloMessage msg; + msg.client_id = "test-client"; + msg.name = "Test Client"; + msg.version = 1; + msg.supported_roles.push_back(SendspinRole::PLAYER); + msg.supported_roles.push_back(SendspinRole::ANNOUNCEMENT); + + AnnouncementSupportObject support; + support.supported_formats.push_back({SendspinCodecFormat::PCM, 1, 48000, 16}); + support.supported_formats.push_back({SendspinCodecFormat::OPUS, 1, 48000, 16}); + support.buffer_capacity = 65536; + msg.announcement_v1_support = support; + + const std::string json = format_client_hello_message(&msg); + + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(json, doc, root)); + + // The role id is listed and the support object is keyed by the versioned alias + bool found_role = false; + for (JsonVariantConst role : root["payload"]["supported_roles"].as()) { + if (role.as() == "announcement@v1") { + found_role = true; + } + } + EXPECT_TRUE(found_role); + + JsonObjectConst support_json = root["payload"]["announcement@v1_support"]; + ASSERT_FALSE(support_json.isNull()); + EXPECT_EQ(support_json["buffer_capacity"].as(), 65536U); + JsonArrayConst formats = support_json["supported_formats"]; + ASSERT_EQ(formats.size(), 2U); + EXPECT_STREQ(formats[0]["codec"].as(), "pcm"); + EXPECT_EQ(formats[0]["channels"].as(), 1); + EXPECT_EQ(formats[0]["sample_rate"].as(), 48000); + EXPECT_EQ(formats[0]["bit_depth"].as(), 16); + EXPECT_STREQ(formats[1]["codec"].as(), "opus"); + + // No supported_commands field exists for the announcement role + EXPECT_TRUE(support_json["supported_commands"].isNull()); +} + +// A hello without announcement support must not emit the key at all +TEST(AnnouncementRole, HelloWithoutSupportOmitsKey) { + ClientHelloMessage msg; + msg.client_id = "test-client"; + msg.name = "Test Client"; + msg.version = 1; + + const std::string json = format_client_hello_message(&msg); + + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(json, doc, root)); + EXPECT_TRUE(root["payload"]["announcement@v1_support"].isNull()); +} + +// ============================================================================ +// stream/start announcement object parsing +// ============================================================================ + +TEST(AnnouncementRole, StreamStartParsesAnnouncementObject) { + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ + "server_transmitted":123, + "announcement":{"codec":"opus","sample_rate":48000,"channels":1,"bit_depth":16, + "codec_header":"aGVhZGVy","start_timestamp":1700000000000000, + "media_duck_db":20,"duck_ramp_ms":250,"volume":60, + "override_mute":true}}})", + doc, root)); + + StreamStartMessage stream_msg; + ASSERT_TRUE(process_stream_start_message(root, &stream_msg)); + ASSERT_TRUE(stream_msg.announcement.has_value()); + EXPECT_FALSE(stream_msg.player.has_value()); + + const ServerAnnouncementStreamObject& announcement = stream_msg.announcement.value(); + EXPECT_TRUE(announcement.is_complete()); + EXPECT_EQ(announcement.format.codec.value(), SendspinCodecFormat::OPUS); + EXPECT_EQ(announcement.format.sample_rate.value(), 48000U); + EXPECT_EQ(announcement.format.channels.value(), 1); + EXPECT_EQ(announcement.format.bit_depth.value(), 16); + EXPECT_EQ(announcement.format.codec_header.value(), "aGVhZGVy"); + EXPECT_EQ(announcement.start_timestamp, 1700000000000000LL); + EXPECT_EQ(announcement.media_duck_db, 20); + EXPECT_EQ(announcement.duck_ramp_ms, 250); + ASSERT_TRUE(announcement.volume.has_value()); + EXPECT_EQ(announcement.volume.value(), 60); + EXPECT_TRUE(announcement.override_mute); +} + +// Optional fields carry defined defaults, and both stream objects can coexist. override_mute +// defaults to false and volume to unset when absent. +TEST(AnnouncementRole, StreamStartDefaultsAndCoexistence) { + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ + "player":{"codec":"flac","sample_rate":48000,"channels":2,"bit_depth":16, + "codec_header":"Zmxh"}, + "announcement":{"codec":"pcm","sample_rate":16000,"channels":1,"bit_depth":16, + "start_timestamp":42}}})", + doc, root)); + + StreamStartMessage stream_msg; + ASSERT_TRUE(process_stream_start_message(root, &stream_msg)); + ASSERT_TRUE(stream_msg.player.has_value()); + ASSERT_TRUE(stream_msg.announcement.has_value()); + + const ServerAnnouncementStreamObject& announcement = stream_msg.announcement.value(); + EXPECT_EQ(announcement.start_timestamp, 42); + EXPECT_EQ(announcement.media_duck_db, 0); // default: no ducking + EXPECT_EQ(announcement.duck_ramp_ms, 100); // default ramp + EXPECT_FALSE(announcement.volume.has_value()); // default: follow master volume + EXPECT_FALSE(announcement.override_mute); // default: silenced while muted +} + +// Out-of-range duck/volume values and a non-boolean override_mute are dropped, leaving defaults +TEST(AnnouncementRole, StreamStartRejectsOutOfRangeFields) { + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ + "announcement":{"codec":"pcm","sample_rate":16000,"channels":1,"bit_depth":16, + "start_timestamp":42,"media_duck_db":51,"duck_ramp_ms":2001, + "volume":101,"override_mute":1}}})", + doc, root)); + + StreamStartMessage stream_msg; + ASSERT_TRUE(process_stream_start_message(root, &stream_msg)); + ASSERT_TRUE(stream_msg.announcement.has_value()); + + const ServerAnnouncementStreamObject& announcement = stream_msg.announcement.value(); + EXPECT_EQ(announcement.media_duck_db, 0); + EXPECT_EQ(announcement.duck_ramp_ms, 100); + EXPECT_FALSE(announcement.volume.has_value()); + EXPECT_FALSE(announcement.override_mute); // integer 1 is not a JSON boolean, so dropped +} + +// start_timestamp is required: an otherwise-complete announcement object without it fails the +// whole stream/start parse. +TEST(AnnouncementRole, StreamStartRequiresStartTimestamp) { + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ + "announcement":{"codec":"pcm","sample_rate":16000,"channels":1,"bit_depth":16}}})", + doc, root)); + + StreamStartMessage stream_msg; + EXPECT_FALSE(process_stream_start_message(root, &stream_msg)); +} + +// An announcement object missing required codec fields fails the whole stream/start parse, +// matching the player object's strictness +TEST(AnnouncementRole, StreamStartRejectsIncompleteAnnouncement) { + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ + "announcement":{"codec":"pcm","sample_rate":16000}}})", + doc, root)); + + StreamStartMessage stream_msg; + EXPECT_FALSE(process_stream_start_message(root, &stream_msg)); +} + +// ============================================================================ +// Re-sent stream/start config update (main-thread event drain) +// ============================================================================ + +// A stream/start arriving while an announcement is active is a config update: the role adopts +// the new duck/volume/override_mute params and re-invokes on_announcement_start, without ending +// the stream or flipping the playing state. +TEST(AnnouncementRole, ResentStreamStartUpdatesConfigInPlace) { + auto impl = make_announcement_impl(); + RecordingListener listener; + impl->listener = &listener; + + // Simulate an already-playing announcement stream. + impl->stream_active = true; + impl->announcement_playing = true; + + ServerAnnouncementStreamObject updated{}; + updated.format.codec = SendspinCodecFormat::PCM; + updated.format.sample_rate = 48000; + updated.format.channels = 1; + updated.format.bit_depth = 16; + updated.start_timestamp = 99; + updated.media_duck_db = 30; + updated.volume = 80; + updated.override_mute = true; + impl->event_state->stream_params_slot.write(updated); + impl->on_stream_ring_event(AnnouncementStreamCallbackType::CONFIG_UPDATE); + + impl->drain_events(); + + // The embedder is re-notified with the new policy; no end, no restart, no state change. + EXPECT_EQ(listener.start_count, 1); + EXPECT_EQ(listener.end_count, 0); + EXPECT_EQ(listener.last_params.media_duck_db, 30); + ASSERT_TRUE(listener.last_params.volume.has_value()); + EXPECT_EQ(listener.last_params.volume.value(), 80); + EXPECT_TRUE(listener.last_params.override_mute); + EXPECT_TRUE(impl->announcement_playing); + EXPECT_EQ(impl->current_stream_params.media_duck_db, 30); +} + +// ============================================================================ +// stream/end termination: graceful drain vs. non-graceful discard +// ============================================================================ + +// A graceful server stream/end must play out any buffered announcement audio before ending: the +// buffered tail is written to the sink, and only then does on_announcement_end() fire (releasing +// the ducking) and the role report idle. The listener gate holds the task inside its first write +// so several chunks are still buffered in the ring when the stream/end arrives. +TEST(AnnouncementRole, StreamEndDrainsBufferedAudioBeforeRelease) { + auto impl = make_announcement_impl(); + DrainRecordingListener listener; + listener.gate_enabled = true; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + + impl->handle_stream_start(make_pcm_stream_params()); + // Fire on_announcement_start (duck applied) and ack the start so the task begins playback. + ASSERT_TRUE(pump_until(*impl, [&] { return listener.starts() >= 1; }, POSITIVE_TIMEOUT)); + + // Feed several chunks; the first blocks in the sink, leaving the rest buffered in the ring. + std::vector expected; + constexpr uint8_t NUM_CHUNKS = 5; + for (uint8_t i = 0; i < NUM_CHUNKS; ++i) { + std::vector chunk = make_pcm_chunk(static_cast('A' + i)); + expected.insert(expected.end(), chunk.begin(), chunk.end()); + impl->handle_binary(chunk.data(), chunk.size()); + } + ASSERT_TRUE(listener.wait_first_write(POSITIVE_TIMEOUT)); + + // Graceful stream/end arrives while chunks are still buffered, then the sink unblocks. + impl->handle_stream_end(); + listener.release_gate(); + + ASSERT_TRUE(pump_until(*impl, [&] { return listener.ends() >= 1; }, POSITIVE_TIMEOUT)); + + // Every fed byte was played out in order (nothing discarded), and the duck released only + // after the full tail had been written. + EXPECT_EQ(listener.written_snapshot(), expected); + EXPECT_EQ(listener.ends(), 1); + EXPECT_EQ(listener.end_bytes(), expected.size()); + EXPECT_FALSE(impl->announcement_playing); +} + +// A non-graceful end (connection teardown via cleanup(), i.e. transport loss) must discard the +// buffered tail and release the ducking immediately -- the opposite of the graceful drain above. +TEST(AnnouncementRole, TransportLossDiscardsBufferedAudio) { + auto impl = make_announcement_impl(); + DrainRecordingListener listener; + listener.gate_enabled = true; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + + impl->handle_stream_start(make_pcm_stream_params()); + ASSERT_TRUE(pump_until(*impl, [&] { return listener.starts() >= 1; }, POSITIVE_TIMEOUT)); + + constexpr uint8_t NUM_CHUNKS = 5; + constexpr size_t CHUNK_BYTES = 128; + for (uint8_t i = 0; i < NUM_CHUNKS; ++i) { + std::vector chunk = make_pcm_chunk(static_cast('A' + i)); + impl->handle_binary(chunk.data(), chunk.size()); + } + ASSERT_TRUE(listener.wait_first_write(POSITIVE_TIMEOUT)); + + // Transport loss: discard whatever is buffered and end now. + impl->cleanup(); + listener.release_gate(); + + ASSERT_TRUE(pump_until(*impl, [&] { return listener.ends() >= 1; }, POSITIVE_TIMEOUT)); + + // The buffered tail was dropped rather than played out: at most the chunk already in flight + // reached the sink, never the whole stream. The duck released immediately (via the STREAM_END + // event), not after a drain. + EXPECT_LT(listener.written_snapshot().size(), NUM_CHUNKS * CHUNK_BYTES); + EXPECT_LE(listener.written_snapshot().size(), CHUNK_BYTES); + EXPECT_EQ(listener.ends(), 1); + EXPECT_FALSE(impl->announcement_playing); +} + +// ============================================================================ +// client/state announcement object serialization +// ============================================================================ + +TEST(AnnouncementRole, ClientStateSerializesAnnouncementObject) { + ClientStateMessage msg; + msg.state = SendspinClientState::SYNCHRONIZED; + + ClientAnnouncementStateObject announcement_state; + announcement_state.playing = true; + announcement_state.required_lead_time_ms = 500; + msg.announcement = announcement_state; + + const std::string json = format_client_state_message(&msg); + + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(json, doc, root)); + EXPECT_STREQ(root["payload"]["announcement"]["state"].as(), "playing"); + EXPECT_EQ(root["payload"]["announcement"]["required_lead_time_ms"].as(), 500); + + // And the idle transition + msg.announcement->playing = false; + msg.announcement->required_lead_time_ms.reset(); + const std::string idle_json = format_client_state_message(&msg); + ASSERT_TRUE(parse(idle_json, doc, root)); + EXPECT_STREQ(root["payload"]["announcement"]["state"].as(), "idle"); + EXPECT_TRUE(root["payload"]["announcement"]["required_lead_time_ms"].isNull()); +} + +// An all-empty announcement state emits no announcement key at all +TEST(AnnouncementRole, ClientStateOmitsEmptyAnnouncementObject) { + ClientStateMessage msg; + msg.state = SendspinClientState::SYNCHRONIZED; + msg.announcement = ClientAnnouncementStateObject{}; + + const std::string json = format_client_state_message(&msg); + + JsonDocument doc; + JsonObject root; + ASSERT_TRUE(parse(json, doc, root)); + EXPECT_TRUE(root["payload"]["announcement"].isNull()); +} diff --git a/tests/test_connection_lifecycle.cpp b/tests/test_connection_lifecycle.cpp index 44811c35..99c125d5 100644 --- a/tests/test_connection_lifecycle.cpp +++ b/tests/test_connection_lifecycle.cpp @@ -20,6 +20,7 @@ // WebSocket upgrade; raw-TCP junk is closed inside the transport layer and never occupies a slot). #include "connection_manager.h" // fnv1_hash, resolve_liveness_timeout_ms +#include "sendspin/announcement_role.h" #include "sendspin/client.h" #include "sendspin/config.h" #include "test_support.h" @@ -61,6 +62,7 @@ constexpr uint16_t ADMIT_TEST_PORT = 18982; constexpr uint16_t LIVENESS_TEST_PORT = 18983; constexpr uint16_t LIVENESS_CONTROL_PORT = 18984; constexpr uint16_t LIVENESS_DISABLED_PORT = 18985; +constexpr uint16_t ANNOUNCEMENT_TEST_PORT = 18991; class TestPersistenceProvider : public SendspinPersistenceProvider { public: @@ -236,6 +238,38 @@ class DelayProxy { bool ok_{false}; }; +// Counts the announcement lifecycle callbacks delivered to the embedder. on_announcement_start / +// on_announcement_end fire on the main loop (the pump thread), on_announcement_write on the task +// thread; atomics keep the test thread's reads safe against the latter. +class AnnouncementRoutingListener : public AnnouncementRoleListener { +public: + size_t on_announcement_write(uint8_t* /*data*/, size_t length, + uint32_t /*timeout_ms*/) override { + return length; // accept everything so the task is never sink-blocked + } + void on_announcement_start(const ServerAnnouncementStreamObject& /*params*/) override { + ++this->starts_; + } + void on_announcement_end() override { + ++this->ends_; + } + int starts() const { + return this->starts_.load(); + } + int ends() const { + return this->ends_.load(); + } + +private: + std::atomic starts_{0}; + std::atomic ends_{0}; +}; + +std::string announcement_stream_start_json() { + return R"({"type":"stream/start","payload":{"announcement":{"codec":"pcm",)" + R"("sample_rate":48000,"channels":1,"bit_depth":16,"start_timestamp":0}}})"; +} + } // namespace // A raw TCP probe (port scan / health check) held open against the client's WS server must not @@ -574,3 +608,41 @@ TEST(ConnectionLifecycle, DisabledLivenessKeepsSilentPeer) { client.disconnect(SendspinGoodbyeReason::SHUTDOWN); pump_for(client, 100); } + +// The announcement stream is independent of the media streams' stream/end. A bare stream/end +// (omitted roles) must leave an active announcement playing and ducked; only a stream/end that +// explicitly names the "announcement" role ends it. Exercises the routing in +// SendspinClient::process_json_message(). +TEST(ConnectionLifecycle, BareStreamEndKeepsAnnouncementExplicitEndsIt) { + TestNetworkProvider network; + SendspinClient client(make_config(ANNOUNCEMENT_TEST_PORT)); + client.set_network_provider(&network); + + AnnouncementRoutingListener listener; + AnnouncementRoleConfig announcement_config; + announcement_config.audio_formats.push_back({SendspinCodecFormat::PCM, 1, 48000, 16}); + client.add_announcement(announcement_config).set_listener(&listener); + + ASSERT_TRUE(client.start()); + client.loop(); // First tick binds the WS server + + FakeServer server(server_url(ANNOUNCEMENT_TEST_PORT), "announce-server"); + pump_until(client, [&] { return client.is_connected(); }); + + // Start an announcement: the embedder is told to duck (on_announcement_start). + server.send_text(announcement_stream_start_json()); + pump_until(client, [&] { return listener.starts() >= 1; }); + + // A bare stream/end must not touch the announcement: no on_announcement_end, duck stays. The + // 500 ms window is well under the task's 5 s stall guard, so nothing else can end the stream. + server.send_text(R"({"type":"stream/end","payload":{}})"); + pump_for(client, 500); + EXPECT_EQ(listener.ends(), 0); + + // A stream/end naming the announcement role ends it (draining the empty buffer immediately). + server.send_text(R"({"type":"stream/end","payload":{"roles":["announcement"]}})"); + pump_until(client, [&] { return listener.ends() >= 1; }); + + client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + pump_for(client, 100); +} From b58f2f6953587d302bc3112bb45f01cc8738105e Mon Sep 17 00:00:00 2001 From: marcelveldt Date: Wed, 16 Sep 2026 20:39:40 +0200 Subject: [PATCH 2/2] Drop override_mute and the media_duck_db cap --- include/sendspin/announcement_role.h | 19 ++++++++----------- src/announcement_role.cpp | 14 +++++++------- src/protocol.cpp | 11 +++-------- tests/test_announcement_role.cpp | 28 ++++++++++++---------------- 4 files changed, 30 insertions(+), 42 deletions(-) diff --git a/include/sendspin/announcement_role.h b/include/sendspin/announcement_role.h index 9e28fc19..4f68be90 100644 --- a/include/sendspin/announcement_role.h +++ b/include/sendspin/announcement_role.h @@ -51,9 +51,10 @@ struct ServerAnnouncementStreamObject { /// multi-speaker start. Required. int64_t start_timestamp{0}; - /// Reduction in decibel (0-50) to apply to this client's own media output while the - /// announcement stream is active. 0 means no ducking. - uint8_t media_duck_db{0}; + /// Non-negative reduction in decibel to apply to this client's own media output while the + /// announcement stream is active. 0 (the default) means no ducking; a large value (e.g. 200) + /// silences the media. + uint16_t media_duck_db{0}; /// Ramp duration in milliseconds (0-2000) for both applying and releasing the ducking. uint16_t duck_ramp_ms{DEFAULT_DUCK_RAMP_MS}; @@ -63,9 +64,6 @@ struct ServerAnnouncementStreamObject { /// the announcement follows the current master volume. std::optional volume{}; - /// When true, a muted client still renders the announcement while its media stays muted. - bool override_mute{false}; - bool is_complete() const { return this->format.is_complete(); } @@ -94,11 +92,10 @@ class AnnouncementRoleListener { /// @brief Called when an announcement stream starts. Fires on the main loop thread /// /// The embedder applies the ducking policy here (e.g. duck the media pipeline by - /// `params.media_duck_db` over `params.duck_ramp_ms`) and honors `params.override_mute` and - /// `params.volume`. Also fires when the server re-sends `stream/start` to update the active - /// announcement's config (duck level, volume, or override_mute): the embedder re-applies the - /// updated policy, ramping a new `media_duck_db` from the current gain, without any - /// intervening on_announcement_end(). + /// `params.media_duck_db` over `params.duck_ramp_ms`) and honors `params.volume`. Also fires + /// when the server re-sends `stream/start` to update the active announcement's config (duck + /// level or volume): the embedder re-applies the updated policy, ramping a new + /// `media_duck_db` from the current gain, without any intervening on_announcement_end(). virtual void on_announcement_start(const ServerAnnouncementStreamObject& /*params*/) {} /// @brief Called when the announcement stream ends (normally, aborted, or on transport diff --git a/src/announcement_role.cpp b/src/announcement_role.cpp index d5f788e9..8cf995e1 100644 --- a/src/announcement_role.cpp +++ b/src/announcement_role.cpp @@ -161,10 +161,10 @@ void AnnouncementRole::Impl::handle_stream_start( } // A stream/start arriving while the announcement is already playing is a configuration update - // (duck level, volume, or override_mute), not a new clip: update the params in place and let - // the main thread re-apply the policy via on_announcement_start, without a codec header, a - // buffer clear, or a task restart. Replacing a clip is done by the server with stream/end - // then a fresh stream/start. + // (duck level or volume), not a new clip: update the params in place and let the main thread + // re-apply the policy via on_announcement_start, without a codec header, a buffer clear, or a + // task restart. Replacing a clip is done by the server with stream/end then a fresh + // stream/start. if (this->task->is_running()) { this->event_state->stream_params_slot.write(announcement_obj); this->enqueue_stream_event(AnnouncementStreamCallbackType::CONFIG_UPDATE); @@ -295,9 +295,9 @@ void AnnouncementRole::Impl::drain_events() { break; } case AnnouncementStreamCallbackType::CONFIG_UPDATE: { - // Re-sent stream/start on an active stream: adopt the new duck/volume/override_mute - // params and re-apply the policy through on_announcement_start. The task keeps - // playing; no state change and no task signal. + // Re-sent stream/start on an active stream: adopt the new duck/volume params and + // re-apply the policy through on_announcement_start. The task keeps playing; no + // state change and no task signal. ServerAnnouncementStreamObject stream_params; if (this->event_state->stream_params_slot.take(stream_params)) { this->current_stream_params = std::move(stream_params); diff --git a/src/protocol.cpp b/src/protocol.cpp index d09f6c42..75f4257e 100644 --- a/src/protocol.cpp +++ b/src/protocol.cpp @@ -34,9 +34,6 @@ static const char* const TAG = "sendspin.protocol"; /// @brief Protocol maximum for volume fields (volume is 0-100 on the wire). static constexpr uint8_t VOLUME_MAX = 100; -/// @brief Protocol maximum for the announcement media_duck_db field (0-50 dB on the wire). -static constexpr uint8_t MEDIA_DUCK_DB_MAX = 50; - /// @brief Protocol maximum for the announcement duck_ramp_ms field (0-2000 ms on the wire). static constexpr uint16_t DUCK_RAMP_MS_MAX = 2000; @@ -712,8 +709,9 @@ bool process_stream_start_message(JsonObject root, StreamStartMessage* stream_ms return false; } announcement_obj.start_timestamp = announcement_json["start_timestamp"].as(); - if (auto v = read_uint_field(announcement_json["media_duck_db"], - "media_duck_db", 0, MEDIA_DUCK_DB_MAX)) { + // media_duck_db is uncapped by the protocol: a large value simply silences the media. + if (auto v = read_uint_field(announcement_json["media_duck_db"], + "media_duck_db")) { announcement_obj.media_duck_db = *v; } if (auto v = read_uint_field(announcement_json["duck_ramp_ms"], @@ -724,9 +722,6 @@ bool process_stream_start_message(JsonObject root, StreamStartMessage* stream_ms VOLUME_MAX)) { announcement_obj.volume = v; } - if (auto v = read_bool_field(announcement_json["override_mute"], "override_mute")) { - announcement_obj.override_mute = *v; - } stream_msg->announcement = std::move(announcement_obj); } else { return false; diff --git a/tests/test_announcement_role.cpp b/tests/test_announcement_role.cpp index d64cf64e..f9f61f1c 100644 --- a/tests/test_announcement_role.cpp +++ b/tests/test_announcement_role.cpp @@ -293,8 +293,7 @@ TEST(AnnouncementRole, StreamStartParsesAnnouncementObject) { "server_transmitted":123, "announcement":{"codec":"opus","sample_rate":48000,"channels":1,"bit_depth":16, "codec_header":"aGVhZGVy","start_timestamp":1700000000000000, - "media_duck_db":20,"duck_ramp_ms":250,"volume":60, - "override_mute":true}}})", + "media_duck_db":200,"duck_ramp_ms":250,"volume":60}}})", doc, root)); StreamStartMessage stream_msg; @@ -310,15 +309,15 @@ TEST(AnnouncementRole, StreamStartParsesAnnouncementObject) { EXPECT_EQ(announcement.format.bit_depth.value(), 16); EXPECT_EQ(announcement.format.codec_header.value(), "aGVhZGVy"); EXPECT_EQ(announcement.start_timestamp, 1700000000000000LL); - EXPECT_EQ(announcement.media_duck_db, 20); + // media_duck_db is uncapped: a large value that silences the media round-trips unchanged + EXPECT_EQ(announcement.media_duck_db, 200); EXPECT_EQ(announcement.duck_ramp_ms, 250); ASSERT_TRUE(announcement.volume.has_value()); EXPECT_EQ(announcement.volume.value(), 60); - EXPECT_TRUE(announcement.override_mute); } -// Optional fields carry defined defaults, and both stream objects can coexist. override_mute -// defaults to false and volume to unset when absent. +// Optional fields carry defined defaults, and both stream objects can coexist. volume defaults to +// unset when absent. TEST(AnnouncementRole, StreamStartDefaultsAndCoexistence) { JsonDocument doc; JsonObject root; @@ -339,17 +338,17 @@ TEST(AnnouncementRole, StreamStartDefaultsAndCoexistence) { EXPECT_EQ(announcement.media_duck_db, 0); // default: no ducking EXPECT_EQ(announcement.duck_ramp_ms, 100); // default ramp EXPECT_FALSE(announcement.volume.has_value()); // default: follow master volume - EXPECT_FALSE(announcement.override_mute); // default: silenced while muted } -// Out-of-range duck/volume values and a non-boolean override_mute are dropped, leaving defaults +// Out-of-range ramp and volume values are dropped, leaving defaults. media_duck_db has no upper +// bound, so a large value is kept. TEST(AnnouncementRole, StreamStartRejectsOutOfRangeFields) { JsonDocument doc; JsonObject root; ASSERT_TRUE(parse(R"({"type":"stream/start","payload":{ "announcement":{"codec":"pcm","sample_rate":16000,"channels":1,"bit_depth":16, - "start_timestamp":42,"media_duck_db":51,"duck_ramp_ms":2001, - "volume":101,"override_mute":1}}})", + "start_timestamp":42,"media_duck_db":200,"duck_ramp_ms":2001, + "volume":101}}})", doc, root)); StreamStartMessage stream_msg; @@ -357,10 +356,9 @@ TEST(AnnouncementRole, StreamStartRejectsOutOfRangeFields) { ASSERT_TRUE(stream_msg.announcement.has_value()); const ServerAnnouncementStreamObject& announcement = stream_msg.announcement.value(); - EXPECT_EQ(announcement.media_duck_db, 0); + EXPECT_EQ(announcement.media_duck_db, 200); EXPECT_EQ(announcement.duck_ramp_ms, 100); EXPECT_FALSE(announcement.volume.has_value()); - EXPECT_FALSE(announcement.override_mute); // integer 1 is not a JSON boolean, so dropped } // start_timestamp is required: an otherwise-complete announcement object without it fails the @@ -394,8 +392,8 @@ TEST(AnnouncementRole, StreamStartRejectsIncompleteAnnouncement) { // ============================================================================ // A stream/start arriving while an announcement is active is a config update: the role adopts -// the new duck/volume/override_mute params and re-invokes on_announcement_start, without ending -// the stream or flipping the playing state. +// the new duck/volume params and re-invokes on_announcement_start, without ending the stream or +// flipping the playing state. TEST(AnnouncementRole, ResentStreamStartUpdatesConfigInPlace) { auto impl = make_announcement_impl(); RecordingListener listener; @@ -413,7 +411,6 @@ TEST(AnnouncementRole, ResentStreamStartUpdatesConfigInPlace) { updated.start_timestamp = 99; updated.media_duck_db = 30; updated.volume = 80; - updated.override_mute = true; impl->event_state->stream_params_slot.write(updated); impl->on_stream_ring_event(AnnouncementStreamCallbackType::CONFIG_UPDATE); @@ -425,7 +422,6 @@ TEST(AnnouncementRole, ResentStreamStartUpdatesConfigInPlace) { EXPECT_EQ(listener.last_params.media_duck_db, 30); ASSERT_TRUE(listener.last_params.volume.has_value()); EXPECT_EQ(listener.last_params.volume.value(), 80); - EXPECT_TRUE(listener.last_params.override_mute); EXPECT_TRUE(impl->announcement_playing); EXPECT_EQ(impl->current_stream_params.media_duck_db, 30); }