diff --git a/CMakeLists.txt b/CMakeLists.txt index c235bea..c03a97e 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 6c72468..88a7f63 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 d32f30c..1556721 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 0000000..1c4c5e3 --- /dev/null +++ b/include/sendspin/announcement_role.h @@ -0,0 +1,164 @@ +// 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{}; + + /// 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{}; + + 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(), +/// on_announcement_end(), and on_announcement_clear() 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`). + 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 Called when the server clears the announcement stream (replace). Fires on the + /// main loop thread. The embedder drops any announcement audio it has buffered downstream; + /// the stream and the ducking state stay active + virtual void on_announcement_clear() {} +}; + +/** + * @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) their first chunk's timestamp - which is + * also what aligns a coordinated multi-speaker announcement scheduled by the server - and are + * 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 5f5fdf2..67033da 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 @@ -258,10 +261,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 @@ -465,6 +485,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 89977c6..45a4f05 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -157,6 +157,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 0000000..62d39a6 --- /dev/null +++ b/src/announcement_role.cpp @@ -0,0 +1,383 @@ +// 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"; + +/// @brief Size of the big-endian 64-bit timestamp at the start of announcement binary messages. +static constexpr size_t BINARY_TIMESTAMP_SIZE = 8; +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; + +/// @brief Swaps bytes of a big-endian 64-bit value to host byte order. +static int64_t be64_to_host(const uint8_t* bytes) { + uint64_t val = 0; + for (int i = 0; i < 8; ++i) { + val = (val << 8) | bytes[i]; + } + return static_cast(val); +} + +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; + } + if (len < BINARY_TIMESTAMP_SIZE) { + SS_LOGW(TAG, "Binary message too short for timestamp"); + return; + } + int64_t timestamp = be64_to_host(data); + // Announcement 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 + BINARY_TIMESTAMP_SIZE, len - BINARY_TIMESTAMP_SIZE, + timestamp, 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; + } + + bool header_sent = false; + const ServerPlayerStreamObject& format = announcement_obj.format; + + 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), 0, 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(), 0, + 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 { + this->task->signal_stream_end(); + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_END); +} + +void AnnouncementRole::Impl::handle_stream_clear() const { + // The server replaces the announcement audio: discard what is buffered, keep the stream + // and the ducking state active. The chunks that follow are the replacement clip. + this->task->signal_stream_clear(); + this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_CLEARED); +} + +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: { + 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::STREAM_CLEARED: { + if (this->listener && this->stream_active) { + this->listener->on_announcement_clear(); + } + break; + } + case AnnouncementStreamCallbackType::OUTPUT_STARTED: { + if (!this->announcement_playing) { + this->announcement_playing = true; + state_changed = true; + } + break; + } + case AnnouncementStreamCallbackType::OUTPUT_FINISHED: { + // Local completion (drained after a server end, or the stall guard fired + // without one): release the ducking if the server end has not already + 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", "STREAM_CLEARED", + "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 0000000..b2a930b --- /dev/null +++ b/src/announcement_role_impl.h @@ -0,0 +1,117 @@ +// 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 ended (from the network thread or cleanup) + STREAM_CLEARED, // Server cleared buffered announcement audio (from the network thread) + OUTPUT_STARTED, // Task began writing announcement audio to the sink (from the task thread) + OUTPUT_FINISHED, // Task finished draining 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 handle_stream_clear() 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 0000000..5e757ab --- /dev/null +++ b/src/announcement_task.cpp @@ -0,0 +1,413 @@ +// 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_CLEAR | + 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_clear() { + if (!this->is_initialized()) { + return; + } + this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR); +} + +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) { + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END); + this->drain_ring_buffer(); + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START) != 0U) { + 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_CLEAR)) != 0U) { + // Stale commands from a stream that ended while idle + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR); + } + + 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 || + chunk_type == CHUNK_TYPE_STREAM_CLEAR_MARKER) { + // Stale audio from a previous announcement; discard until a codec header arrives + this->encoded_ring_buffer_->return_chunk(entry); + continue; + } + + 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->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; + 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) { + // Stream over (server end or abort): discard any remaining buffered audio + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END); + this->drain_ring_buffer(); + break; + } + if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR) != 0U) { + // Replace: drop what is buffered, keep playing what follows. The clear boundary is + // approximate (chunks racing the drain may be dropped with the old clip), which the + // silence-padded replacement stream tolerates. + this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR); + this->drain_ring_buffer(); + continue; + } + + AudioRingBufferEntry* entry = + this->encoded_ring_buffer_->receive_chunk(ANNOUNCEMENT_CHUNK_RECEIVE_TIMEOUT_MS); + if (entry == nullptr) { + 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) { + if (entry->chunk_type == CHUNK_TYPE_STREAM_CLEAR_MARKER) { + this->encoded_ring_buffer_->return_chunk(entry); + continue; + } + // 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); + } + const int64_t chunk_timestamp = entry->timestamp; + 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 first chunk's timestamp when time sync is + // available, or as soon as possible otherwise. Announcement chunks are never dropped + // for lateness. + this->wait_for_start_time(chunk_timestamp); + if ((this->event_flags_.get() & + (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR)) != 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 | + AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR)) != 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::drain_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 0000000..0bfd3c9 --- /dev/null +++ b/src/announcement_task.h @@ -0,0 +1,165 @@ +// 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), // Signal end of current announcement stream + ANNOUNCEMENT_COMMAND_STREAM_CLEAR = (1 << 2), // Replace: discard buffered announcement audio + 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, the first chunk starts at (or as soon as possible after) its 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 end the current announcement stream. Non-blocking + /// The task drains stale audio from the ring buffer and returns to idle. + /// Thread-safe: may be called from any context. + void signal_stream_end(); + + /// @brief Signals the task that the server cleared the announcement stream (replace). + /// Non-blocking. The task discards currently buffered announcement audio and keeps the + /// stream open for the chunks that follow. + /// Thread-safe: may be called from any context. + void signal_stream_clear(); + + /// @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 Server timestamp for this chunk. + /// @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. 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 first chunk's 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 drain of all chunks currently in the ring buffer + void drain_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_; + + // 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 6efbc00..151505e 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 @@ -86,6 +89,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 @@ -130,6 +136,14 @@ bool SendspinClient::start_server() { } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_) { + if (!this->announcement_->impl_->start()) { + return false; + } + } +#endif + #ifdef SENDSPIN_ENABLE_VISUALIZER if (this->visualizer_) { if (!this->visualizer_->impl_->start()) { @@ -302,6 +316,17 @@ void SendspinClient::loop() { 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; } @@ -337,6 +362,11 @@ void SendspinClient::loop() { 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(); @@ -461,6 +491,17 @@ VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { } #endif +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT +AnnouncementRole& SendspinClient::add_announcement(AnnouncementRoleConfig config) { + if (this->started_) { + SS_LOGW(TAG, "add_announcement() called after start_server()"); + } + this->announcement_ = std::make_unique(std::move(config), this); + this->announcement_->impl_->attach_inbox(this->event_state_->inbox); + return *this->announcement_; +} +#endif + // ============================================================================ // Queries // ============================================================================ @@ -565,6 +606,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(); @@ -655,6 +701,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); } @@ -714,14 +765,22 @@ 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 ALL active streams, including an announcement stream 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 = !end_msg.roles.has_value(); if (end_msg.roles.has_value()) { for (const auto& role : end_msg.roles.value()) { @@ -731,12 +790,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) { @@ -755,15 +816,24 @@ 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)) { + // Omitted roles is a media seek: it clears the player and visualizer streams + // only. The announcement stream is cleared solely when listed explicitly. bool clear_player = !clear_msg.roles.has_value(); bool clear_artwork = !clear_msg.roles.has_value(); bool clear_visualizer = !clear_msg.roles.has_value(); + bool clear_announcement = false; if (clear_msg.roles.has_value()) { for (const auto& role : clear_msg.roles.value()) { @@ -773,12 +843,14 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* clear_artwork = true; } else if (role == "visualizer") { clear_visualizer = true; + } else if (role == "announcement") { + clear_announcement = true; } } } - SS_LOGD(TAG, "Stream clear - player:%d artwork:%d visualizer:%d", clear_player, - clear_artwork, clear_visualizer); + SS_LOGD(TAG, "Stream clear - player:%d artwork:%d visualizer:%d announcement:%d", + clear_player, clear_artwork, clear_visualizer, clear_announcement); #ifdef SENDSPIN_ENABLE_PLAYER if (this->player_ && clear_player) { @@ -797,6 +869,12 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* this->visualizer_->impl_->handle_stream_clear(); } #endif + +#ifdef SENDSPIN_ENABLE_ANNOUNCEMENT + if (this->announcement_ && clear_announcement) { + this->announcement_->impl_->handle_stream_clear(); + } +#endif } break; } @@ -933,6 +1011,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; } @@ -961,6 +1051,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 2f031f0..b3a882e 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 6a5a7bb..57c2fc7 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 @@ -704,6 +710,33 @@ 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; + } + 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; + } + stream_msg->announcement = std::move(announcement_obj); + } else { + return false; + } + } + return true; } @@ -897,6 +930,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; @@ -923,6 +970,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 e02ccc9..60df304 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/state message containing per-role state updates @@ -652,6 +678,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 ac5113c..0e1064a 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -29,6 +29,7 @@ add_executable(sendspin_tests test_inbox.cpp test_visualizer_role.cpp test_artwork_role.cpp + test_announcement_role.cpp ) # Reach the library's private headers (protocol_messages.h, time_filter.h, ...). diff --git a/tests/test_announcement_role.cpp b/tests/test_announcement_role.cpp new file mode 100644 index 0000000..fe24180 --- /dev/null +++ b/tests/test_announcement_role.cpp @@ -0,0 +1,247 @@ +// 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 "protocol_messages.h" +#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; +} + +} // 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","media_duck_db":20,"duck_ramp_ms":250, + "volume":60}}})", + 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.media_duck_db, 20); + EXPECT_EQ(announcement.duck_ramp_ms, 250); + ASSERT_TRUE(announcement.volume.has_value()); + EXPECT_EQ(announcement.volume.value(), 60); +} + +// Duck/volume fields are optional with defined defaults, and both stream objects can coexist +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}}})", + 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.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 +} + +// Out-of-range duck/volume values are dropped, leaving the 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, + "media_duck_db":51,"duck_ramp_ms":2001,"volume":101}}})", + 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()); +} + +// 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)); +} + +// ============================================================================ +// 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()); +}