From 2d30ccf8b6dce1e101e210cf2586da0aa57c6c6b Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Mon, 31 Aug 2026 20:43:40 -0500 Subject: [PATCH] Add an outbound binary send primitive to all transports --- src/connection.h | 18 +++ src/esp/client_connection.cpp | 28 +++++ src/esp/client_connection.h | 7 ++ src/esp/server_connection.cpp | 118 +++++++++++++++++++ src/esp/server_connection.h | 46 +++++++- src/host/client_connection.cpp | 25 ++++ src/host/client_connection.h | 7 ++ src/host/server_connection.cpp | 20 ++++ src/host/server_connection.h | 8 ++ src/platform/types.h | 1 + tests/test_connection_lifecycle.cpp | 177 ++++++++++++++++++++++++++++ 11 files changed, 454 insertions(+), 1 deletion(-) diff --git a/src/connection.h b/src/connection.h index 1c4f4fe..41089c2 100644 --- a/src/connection.h +++ b/src/connection.h @@ -152,6 +152,24 @@ class SendspinConnection : public std::enable_shared_from_thisis_connected()) { + if (cb) { + cb(false); + } + return SsErr::INVALID_STATE; + } + + // esp_websocket_client_send_bin is synchronous in the current task, like the text path + int sent = esp_websocket_client_send_bin(this->client_, reinterpret_cast(data), + static_cast(len), + pdMS_TO_TICKS(WEBSOCKET_SEND_TIMEOUT_MS)); + + bool success = (sent >= 0); + + if (cb) { + cb(success); + } + + if (!success) { + SS_LOGE(TAG, "Failed to send binary message (timeout or error): %d", sent); + return SsErr::FAIL; + } + + return SsErr::OK; +} + bool SendspinClientConnection::send_time_message() { if (!this->is_connected()) { return false; diff --git a/src/esp/client_connection.h b/src/esp/client_connection.h index 3db66c1..a6cbf38 100644 --- a/src/esp/client_connection.h +++ b/src/esp/client_connection.h @@ -90,6 +90,13 @@ class SendspinClientConnection : public SendspinConnection { SsErr send_text_message(const std::string& message, SendCompleteCallback cb, bool allow_before_hello) override; + /// @brief Sends a binary message to the server, synchronously like the text path + /// @param data Pointer to the message bytes. + /// @param len Length of the message in bytes. + /// @param cb Callback invoked inline in the calling thread with the send result. + /// @return SsErr::OK if sent successfully, error code otherwise. + SsErr send_binary_message(const uint8_t* data, size_t len, SendCompleteCallback cb) override; + /// @brief Sends a client/time message, capturing the timestamp just before send /// @return true if the message was sent successfully, false otherwise. bool send_time_message() override; diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index 5a0d504..3cc0244 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -55,6 +55,19 @@ struct SessionLookup { std::weak_ptr conn; }; +/// @brief Once-per-connection identity block for queued binary send work (a reusable +/// SessionLookup: the binary path runs per chunk and must not allocate in steady state) +/// +/// While a work item is queued, `self` keeps the block alive independently of the connection; +/// the worker moves `self` into a local before resolving `conn`, so teardown with work in +/// flight makes the worker a clean no-op. If httpd stops and discards queued work, the engaged +/// `self` cycle leaks this small block (bounded by client shutdowns); the destructor still +/// fails the pending completion. +struct BinarySendLookup { + std::weak_ptr conn; + std::shared_ptr self; +}; + // ============================================================================ // SendspinConnection interface implementation // ============================================================================ @@ -68,6 +81,15 @@ SendspinServerConnection::SendspinServerConnection(httpd_handle_t server, int so } } +SendspinServerConnection::~SendspinServerConnection() { + // A still-queued worker can never touch this connection again (weak_ptr lock fails), so the + // pending completion is failed here; a worker that DID lock blocks destruction until done + if (this->binary_send_in_flight_.load(std::memory_order_acquire) && this->binary_send_cb_) { + SendCompleteCallback pending = std::move(this->binary_send_cb_); + pending(false); + } +} + void SendspinServerConnection::start() { // Time filter is initialized by the hub when it sets up the connection. } @@ -172,6 +194,102 @@ SsErr SendspinServerConnection::send_text_message(const std::string& message, return SsErr::OK; } +SsErr SendspinServerConnection::send_binary_message(const uint8_t* data, size_t len, + SendCompleteCallback on_complete) { + if (!this->is_connected()) { + if (on_complete) { + on_complete(false); + } + return SsErr::INVALID_STATE; + } + + // Single-in-flight slot: a chunk arriving while the previous is still queued is rejected + // and the caller drops it (the spec's stall policy) + if (this->binary_send_in_flight_.exchange(true, std::memory_order_acq_rel)) { + if (on_complete) { + on_complete(false); + } + return SsErr::NOT_FINISHED; + } + + // Grow-only buffer sized by the first payload: chunks are near-constant size, so steady + // state allocates nothing and any growth is loud. SPIRAM-preferred like the receive buffer. + if (this->binary_send_payload_.size() < len) { + bool grown; + if (this->binary_send_payload_.data() == nullptr) { + grown = this->binary_send_payload_.allocate(len, MemoryLocation::PREFER_EXTERNAL); + } else { + SS_LOGW(TAG, "Growing binary send slot %zu -> %zu bytes", + this->binary_send_payload_.size(), len); + grown = this->binary_send_payload_.realloc(len); + } + if (!grown) { + SS_LOGE(TAG, "Failed to allocate %zu bytes for binary send slot", len); + this->binary_send_in_flight_.store(false, std::memory_order_release); + if (on_complete) { + on_complete(false); + } + return SsErr::NO_MEM; + } + } + + std::memcpy(this->binary_send_payload_.data(), data, len); + this->binary_send_len_ = len; + this->binary_send_cb_ = std::move(on_complete); + + if (this->binary_send_lookup_ == nullptr) { + this->binary_send_lookup_ = std::make_shared(); + this->binary_send_lookup_->conn = + std::static_pointer_cast(this->shared_from_this()); + } + // Engage the keep-alive reference for the queued worker (see BinarySendLookup). + this->binary_send_lookup_->self = this->binary_send_lookup_; + + if (httpd_queue_work(this->server_, async_send_binary, this->binary_send_lookup_.get()) != + ESP_OK) { + SS_LOGE(TAG, "httpd_queue_work failed for binary message"); + this->binary_send_lookup_->self.reset(); + SendCompleteCallback pending = std::move(this->binary_send_cb_); + this->binary_send_in_flight_.store(false, std::memory_order_release); + if (pending) { + pending(false); + } + return SsErr::FAIL; + } + return SsErr::OK; +} + +void SendspinServerConnection::async_send_binary(void* arg) { + auto* lookup = static_cast(arg); + // Take the keep-alive back first; a successful lock() then blocks destruction until return + std::shared_ptr keep = std::move(lookup->self); + auto conn = lookup->conn.lock(); + if (conn == nullptr) { + return; // Torn down with work queued: the destructor already failed the completion + } + + bool success = false; + // Same identity and hello gating as async_send_text + if (conn->is_connected() && conn->client_hello_sent_) { + httpd_ws_frame_t ws_pkt; + memset(&ws_pkt, 0, sizeof(httpd_ws_frame_t)); + ws_pkt.payload = conn->binary_send_payload_.data(); + ws_pkt.len = conn->binary_send_len_; + ws_pkt.type = HTTPD_WS_TYPE_BINARY; + success = httpd_ws_send_frame_async(conn->server_, conn->sockfd_, &ws_pkt) == ESP_OK; + } + + // The completion fires on every exit path with a live connection (sent, send failed, gated, + // or already disconnected) — the slot would wedge otherwise. The callback is moved out and + // the slot released before invoking it, so a completion that immediately sends the next + // chunk finds the slot free. + SendCompleteCallback pending = std::move(conn->binary_send_cb_); + conn->binary_send_in_flight_.store(false, std::memory_order_release); + if (pending) { + pending(success); + } +} + void SendspinServerConnection::trigger_close() { // Gate on is_connected(): once close_callback has marked this connection closed, httpd may // recycle the fd onto a freshly-accepted session, and closing by the stale fd would kill the diff --git a/src/esp/server_connection.h b/src/esp/server_connection.h index 2b70b16..db0d052 100644 --- a/src/esp/server_connection.h +++ b/src/esp/server_connection.h @@ -18,13 +18,17 @@ #pragma once #include "connection.h" +#include "platform/memory.h" #include #include #include +#include namespace sendspin { +struct BinarySendLookup; + /** * @brief ESP-IDF HTTP server WebSocket connection representing a single Sendspin server session * @@ -57,7 +61,10 @@ class SendspinServerConnection : public SendspinConnection { /// @param sockfd The socket file descriptor for this connection. SendspinServerConnection(httpd_handle_t server, int sockfd); - ~SendspinServerConnection() override = default; + /// @brief Fails a still-in-flight binary send whose queued worker can never run anymore + /// (teardown half of the send-slot contract: only the worker or this destructor releases + /// the slot, never a caller-side timeout) + ~SendspinServerConnection() override; // ======================================== // SendspinConnection interface implementation @@ -103,6 +110,17 @@ class SendspinServerConnection : public SendspinConnection { SsErr send_text_message(const std::string& message, SendCompleteCallback on_complete, bool allow_before_hello) override; + /// @brief Sends a binary message through the connection's single-in-flight send slot: + /// allocation-free in steady state (per-chunk path), NOT_FINISHED while the previous send + /// is in flight, slot released only by the worker's completion or the destructor + /// @param data Pointer to the message bytes. + /// @param len Length of the message in bytes. + /// @param on_complete Callback invoked with the send result. + /// @return SsErr::OK if queued; SsErr::NOT_FINISHED when the slot is busy; other codes on + /// failure. + SsErr send_binary_message(const uint8_t* data, size_t len, + SendCompleteCallback on_complete) override; + /// @brief Sends a client/time message, stamping the timestamp inside the httpd worker /// /// Schedules a worker job that captures `client_transmitted` and serializes the JSON @@ -152,11 +170,32 @@ class SendspinServerConnection : public SendspinConnection { /// destroyed and freed before the worker returns. static void async_send_time_text(void* arg); + /// @brief httpd_queue_work callback that sends the binary frame held in the send slot + /// @param arg This connection's BinarySendLookup (lifetime protocol at its definition). + static void async_send_binary(void* arg); + + // Struct fields + + /// @brief Binary send slot payload: sized by the first send, grow-only, reused (zero + /// steady-state allocation) + PlatformBuffer binary_send_payload_; + + /// @brief In-flight completion callback, fired by the worker or the destructor + SendCompleteCallback binary_send_cb_; + // Pointer fields /// @brief The httpd server handle (owned by SendspinWsServer) httpd_handle_t server_; + /// @brief Identity block handed to queued binary send work (see BinarySendLookup) + std::shared_ptr binary_send_lookup_; + + // size_t fields + + /// @brief Length of the payload currently held in the binary send slot + size_t binary_send_len_{0}; + // 32-bit fields /// @brief The socket file descriptor for this connection @@ -166,6 +205,11 @@ class SendspinServerConnection : public SendspinConnection { /// @brief Set once the httpd session has closed (see mark_closed()) std::atomic closed_{false}; + + /// @brief True while a binary send occupies the slot (queued or being sent). Written by the + /// sending role thread (acquire the slot) and the httpd worker (release it); the destructor + /// reads it to detect work that never ran. + std::atomic binary_send_in_flight_{false}; }; } // namespace sendspin diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index ee5503c..bf8e9c7 100644 --- a/src/host/client_connection.cpp +++ b/src/host/client_connection.cpp @@ -120,6 +120,31 @@ SsErr SendspinClientConnection::send_text_message(const std::string& message, return SsErr::OK; } +SsErr SendspinClientConnection::send_binary_message(const uint8_t* data, size_t len, + SendCompleteCallback cb) { + if (!this->is_connected()) { + if (cb) { + cb(false); + } + return SsErr::INVALID_STATE; + } + + // IXWebSocket's API forces a std::string copy of the payload; acceptable on host. + auto info = this->ws_->sendBinary(std::string(reinterpret_cast(data), len)); + bool success = info.success; + + if (cb) { + cb(success); + } + + if (!success) { + SS_LOGE(TAG, "Failed to send binary message"); + return SsErr::FAIL; + } + + return SsErr::OK; +} + bool SendspinClientConnection::send_time_message() { if (!this->is_connected()) { return false; diff --git a/src/host/client_connection.h b/src/host/client_connection.h index 7ec672c..1e0b627 100644 --- a/src/host/client_connection.h +++ b/src/host/client_connection.h @@ -76,6 +76,13 @@ class SendspinClientConnection : public SendspinConnection { SsErr send_text_message(const std::string& message, SendCompleteCallback cb, bool allow_before_hello) override; + /// @brief Sends a binary message to the server, synchronously like the text path + /// @param data Pointer to the message bytes. + /// @param len Length of the message in bytes. + /// @param cb Callback invoked inline in the calling thread with the send result. + /// @return SsErr::OK if sent successfully, error code otherwise. + SsErr send_binary_message(const uint8_t* data, size_t len, SendCompleteCallback cb) override; + /// @brief Sends a client/time message, capturing the timestamp synchronously before send /// @return true if the message was sent successfully, false otherwise. bool send_time_message() override; diff --git a/src/host/server_connection.cpp b/src/host/server_connection.cpp index 95f73b2..0b5643a 100644 --- a/src/host/server_connection.cpp +++ b/src/host/server_connection.cpp @@ -83,6 +83,26 @@ SsErr SendspinServerConnection::send_text_message(const std::string& message, return success ? SsErr::OK : SsErr::FAIL; } +SsErr SendspinServerConnection::send_binary_message(const uint8_t* data, size_t len, + SendCompleteCallback on_complete) { + if (!this->is_connected()) { + if (on_complete) { + on_complete(false); + } + return SsErr::INVALID_STATE; + } + + // IXWebSocket's API forces a std::string copy of the payload; acceptable on host. + auto info = this->ws_->sendBinary(std::string(reinterpret_cast(data), len)); + bool success = info.success; + + if (on_complete) { + on_complete(success); + } + + return success ? SsErr::OK : SsErr::FAIL; +} + bool SendspinServerConnection::send_time_message() { if (!this->is_connected()) { return false; diff --git a/src/host/server_connection.h b/src/host/server_connection.h index 0254005..81307d5 100644 --- a/src/host/server_connection.h +++ b/src/host/server_connection.h @@ -77,6 +77,14 @@ class SendspinServerConnection : public SendspinConnection { SsErr send_text_message(const std::string& message, SendCompleteCallback on_complete, bool allow_before_hello) override; + /// @brief Sends a binary message to the connected client, synchronously like the text path + /// @param data Pointer to the message bytes. + /// @param len Length of the message in bytes. + /// @param on_complete Callback invoked inline in the calling thread with the send result. + /// @return SsErr::OK if sent successfully, error code otherwise. + SsErr send_binary_message(const uint8_t* data, size_t len, + SendCompleteCallback on_complete) override; + /// @brief Sends a client/time message, capturing the timestamp synchronously before send /// @return true if the message was sent successfully, false otherwise. bool send_time_message() override; diff --git a/src/platform/types.h b/src/platform/types.h index 9e1eca9..8f60f4c 100644 --- a/src/platform/types.h +++ b/src/platform/types.h @@ -38,6 +38,7 @@ enum class SsErr : int16_t { NOT_FOUND = 0x105, // Resource not found NOT_SUPPORTED = 0x106, // Operation not supported TIMEOUT = 0x107, // Operation timed out + NOT_FINISHED = 0x10C, // Previous operation has not fully completed // Errors (< 0) FAIL = -1, // Generic failure diff --git a/tests/test_connection_lifecycle.cpp b/tests/test_connection_lifecycle.cpp index be2bf82..1e7c4de 100644 --- a/tests/test_connection_lifecycle.cpp +++ b/tests/test_connection_lifecycle.cpp @@ -20,6 +20,9 @@ // WebSocket upgrade; raw-TCP junk is closed inside the transport layer and never occupies a slot). #include "connection_manager.h" // fnv1_hash for the last-played preference +#include "host/client_connection.h" +#include "host/server_connection.h" +#include "protocol_messages.h" #include "sendspin/client.h" #include "sendspin/config.h" #include @@ -36,7 +39,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -56,6 +61,8 @@ constexpr uint16_t EVICT_TEST_PORT = 18972; constexpr uint16_t REJECT_TEST_PORT = 18973; constexpr uint16_t STALL_LISTEN_PORT = 18981; constexpr uint16_t ADMIT_TEST_PORT = 18982; +constexpr uint16_t BINARY_TEST_PORT = 18992; +constexpr uint16_t SERVER_BINARY_TEST_PORT = 18993; std::string server_url(uint16_t port) { return "ws://127.0.0.1:" + std::to_string(port) + "/sendspin"; @@ -598,3 +605,173 @@ TEST(ConnectionLifecycle, FullNurseryOfLivePeersRejectsNewcomer) { EXPECT_FALSE(mute_a.closed()); EXPECT_FALSE(mute_b.closed()); } + +// send_binary_message on the host client transport must deliver exactly one binary WebSocket +// frame carrying the exact payload bytes, reporting success through both the return code and the +// inline completion callback. +TEST(ConnectionLifecycle, ClientBinarySendDeliversExactBytes) { + std::atomic binary_frames{0}; + std::string received; + std::mutex received_mutex; + + ix::WebSocketServer backend(BINARY_TEST_PORT, "127.0.0.1"); + backend.setOnConnectionCallback([&](const std::weak_ptr& weak_ws, + const std::shared_ptr& /*state*/) { + auto ws = weak_ws.lock(); + if (!ws) { + return; + } + ws->setOnMessageCallback([&](const ix::WebSocketMessagePtr& msg) { + if (msg->type == ix::WebSocketMessageType::Message && msg->binary) { + std::lock_guard lock(received_mutex); + received = msg->str; + binary_frames.fetch_add(1); + } + }); + }); + ASSERT_TRUE(backend.listen().first); + backend.start(); + + SendspinClientConnection conn(server_url(BINARY_TEST_PORT)); + std::atomic connected{false}; + conn.on_connected_cb = [&](SendspinConnection*) { connected.store(true); }; + conn.start(); + { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!connected.load() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + } + ASSERT_TRUE(connected.load()); + + // Payload shaped like a source audio chunk (type byte, BE64 capture timestamp, frame data), + // with bytes chosen to catch truncation and sign mangling. + uint8_t payload[1 + BINARY_TIMESTAMP_SIZE + 4]; + payload[0] = SENDSPIN_BINARY_SOURCE_AUDIO; + host_to_be64(-123456789, payload + 1); + payload[1 + BINARY_TIMESTAMP_SIZE + 0] = 0x00; + payload[1 + BINARY_TIMESTAMP_SIZE + 1] = 0xFF; + payload[1 + BINARY_TIMESTAMP_SIZE + 2] = 0x7F; + payload[1 + BINARY_TIMESTAMP_SIZE + 3] = 0x80; + + bool cb_fired = false; + bool cb_success = false; + EXPECT_EQ(conn.send_binary_message(payload, sizeof(payload), + [&](bool ok) { + cb_fired = true; + cb_success = ok; + }), + SsErr::OK); + // Synchronous transport: the completion fires inline in the calling thread. + EXPECT_TRUE(cb_fired); + EXPECT_TRUE(cb_success); + + { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (binary_frames.load() == 0 && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + } + ASSERT_EQ(binary_frames.load(), 1); + { + std::lock_guard lock(received_mutex); + ASSERT_EQ(received.size(), sizeof(payload)); + EXPECT_EQ(0, std::memcmp(received.data(), payload, sizeof(payload))); + } + + backend.stop(); +} + +// Mirror of ClientBinarySendDeliversExactBytes for the host server transport — the transport a +// source device uses when the Sendspin server connects inbound, so its parallel but independent +// send implementation gets its own exact-bytes pin. The library-side connection is constructed +// directly around the accepted IXWebSocket, the way the host ws_server builds it. +TEST(ConnectionLifecycle, ServerBinarySendDeliversExactBytes) { + std::mutex conn_mutex; + std::shared_ptr server_conn; + + ix::WebSocketServer listener(SERVER_BINARY_TEST_PORT, "127.0.0.1"); + listener.setOnConnectionCallback([&](const std::weak_ptr& weak_ws, + const std::shared_ptr& /*state*/) { + auto ws = weak_ws.lock(); + if (!ws) { + return; + } + // IXWebSocket requires a message callback on every accepted socket; inbound frames are + // irrelevant here, only the outbound send path is under test. + ws->setOnMessageCallback([](const ix::WebSocketMessagePtr& /*msg*/) {}); + std::lock_guard lock(conn_mutex); + server_conn = std::make_shared(ws, -1); + }); + ASSERT_TRUE(listener.listen().first); + listener.start(); + + // The peer that receives the frame: an IXWebSocket client capturing binary messages. + std::atomic binary_frames{0}; + std::string received; + std::mutex received_mutex; + ix::WebSocket peer; + peer.setUrl(server_url(SERVER_BINARY_TEST_PORT)); + peer.disableAutomaticReconnection(); + peer.setOnMessageCallback([&](const ix::WebSocketMessagePtr& msg) { + if (msg->type == ix::WebSocketMessageType::Message && msg->binary) { + std::lock_guard lock(received_mutex); + received = msg->str; + binary_frames.fetch_add(1); + } + }); + peer.start(); + + auto conn_ready = [&] { + std::lock_guard lock(conn_mutex); + return server_conn != nullptr && server_conn->is_connected(); + }; + { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!conn_ready() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + } + ASSERT_TRUE(conn_ready()); + std::shared_ptr conn; + { + std::lock_guard lock(conn_mutex); + conn = server_conn; + } + + uint8_t payload[1 + BINARY_TIMESTAMP_SIZE + 4]; + payload[0] = SENDSPIN_BINARY_SOURCE_AUDIO; + host_to_be64(-123456789, payload + 1); + payload[1 + BINARY_TIMESTAMP_SIZE + 0] = 0x00; + payload[1 + BINARY_TIMESTAMP_SIZE + 1] = 0xFF; + payload[1 + BINARY_TIMESTAMP_SIZE + 2] = 0x7F; + payload[1 + BINARY_TIMESTAMP_SIZE + 3] = 0x80; + + bool cb_fired = false; + bool cb_success = false; + EXPECT_EQ(conn->send_binary_message(payload, sizeof(payload), + [&](bool ok) { + cb_fired = true; + cb_success = ok; + }), + SsErr::OK); + // Synchronous transport: the completion fires inline in the calling thread. + EXPECT_TRUE(cb_fired); + EXPECT_TRUE(cb_success); + + { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (binary_frames.load() == 0 && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + } + ASSERT_EQ(binary_frames.load(), 1); + { + std::lock_guard lock(received_mutex); + ASSERT_EQ(received.size(), sizeof(payload)); + EXPECT_EQ(0, std::memcmp(received.data(), payload, sizeof(payload))); + } + + peer.stop(); + listener.stop(); +}