Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions src/connection.h
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,24 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
virtual SsErr send_text_message(const std::string& message, SendCompleteCallback cb,
bool allow_before_hello = false) = 0;

/// @brief Sends a binary message to the server with a completion callback
///
/// Callable from role task threads via ConnectionManager::current_shared(). The payload
/// must stay valid until @p cb fires or the call returns an error (queuing transports copy
/// it first). Binary frames are gated behind the client/hello like the default text path.
///
/// Unlike the text path's best-effort callback, @p cb fires exactly once for EVERY call --
/// on success, every failure path, and connection teardown -- because single-in-flight
/// transports release their send slot from the completion path.
///
/// @param data Pointer to the message bytes (type byte first).
/// @param len Length of the message in bytes.
/// @param cb Callback invoked with the send result.
/// @return SsErr::OK if sent/queued successfully; SsErr::NOT_FINISHED if a previous binary
/// send is still in flight on a single-in-flight transport (the caller treats this
/// as "drop this chunk" and owns logging that drop); other codes on failure.
virtual SsErr send_binary_message(const uint8_t* data, size_t len, SendCompleteCallback cb) = 0;

/// @brief Sends a client/time synchronization message
///
/// The transport implementation captures `client_transmitted` as close to the actual wire
Expand Down
28 changes: 28 additions & 0 deletions src/esp/client_connection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,34 @@ 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;
}

// 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<const char*>(data),
static_cast<int>(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;
Expand Down
7 changes: 7 additions & 0 deletions src/esp/client_connection.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
118 changes: 118 additions & 0 deletions src/esp/server_connection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,19 @@ struct SessionLookup {
std::weak_ptr<SendspinServerConnection> 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<SendspinServerConnection> conn;
std::shared_ptr<BinarySendLookup> self;
};

// ============================================================================
// SendspinConnection interface implementation
// ============================================================================
Expand All @@ -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.
}
Expand Down Expand Up @@ -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<BinarySendLookup>();
this->binary_send_lookup_->conn =
std::static_pointer_cast<SendspinServerConnection>(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<BinarySendLookup*>(arg);
// Take the keep-alive back first; a successful lock() then blocks destruction until return
std::shared_ptr<BinarySendLookup> 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
Expand Down
46 changes: 45 additions & 1 deletion src/esp/server_connection.h
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,17 @@
#pragma once

#include "connection.h"
#include "platform/memory.h"
#include <esp_http_server.h>

#include <atomic>
#include <functional>
#include <memory>

namespace sendspin {

struct BinarySendLookup;

/**
* @brief ESP-IDF HTTP server WebSocket connection representing a single Sendspin server session
*
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<BinarySendLookup> 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
Expand All @@ -166,6 +205,11 @@ class SendspinServerConnection : public SendspinConnection {

/// @brief Set once the httpd session has closed (see mark_closed())
std::atomic<bool> 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<bool> binary_send_in_flight_{false};
};

} // namespace sendspin
25 changes: 25 additions & 0 deletions src/host/client_connection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<const char*>(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;
Expand Down
7 changes: 7 additions & 0 deletions src/host/client_connection.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
20 changes: 20 additions & 0 deletions src/host/server_connection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<const char*>(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;
Expand Down
8 changes: 8 additions & 0 deletions src/host/server_connection.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading