transport.c File Reference

WebSocket transport implementation for ACIP protocol. More...

Go to the source code of this file.

Macros
#define	WEBSOCKET_MESSAGE_QUEUE_SIZE_INCOMING   64
	Maximum incoming message queue size (frames received from peer)
#define	WEBSOCKET_MESSAGE_QUEUE_SIZE_OUTGOING   512
	Maximum outgoing message queue size (frames to send to peer)
#define	WEBSOCKET_RECV_QUEUE_SIZE   WEBSOCKET_MESSAGE_QUEUE_SIZE_INCOMING
#define	WEBSOCKET_SEND_QUEUE_SIZE   WEBSOCKET_MESSAGE_QUEUE_SIZE_OUTGOING

Functions
acip_transport_t *	acip_websocket_client_transport_create (const char *name, const char *url, crypto_context_t *crypto_ctx)
	Create WebSocket client transport.
acip_transport_t *	acip_websocket_server_transport_create (const char *name, struct lws *wsi, crypto_context_t *crypto_ctx)
	Create WebSocket server transport from existing connection.

Detailed Description

WebSocket transport implementation for ACIP protocol.

Implements the acip_transport_t interface for WebSocket connections. Enables browser clients to connect via WebSocket protocol.

ARCHITECTURE:

• Uses libwebsockets for WebSocket protocol handling
• Async libwebsockets callbacks bridge to sync recv() via ringbuffer
• Thread-safe receive queue handles async message arrival
• Same pattern as WebRTC transport for consistency

MESSAGE FLOW:

1. send(): Synchronous write via lws_write()
2. LWS callback: Async write to receive ringbuffer
3. recv(): Blocking read from receive ringbuffer

MEMORY OWNERSHIP:

• Transport OWNS wsi (WebSocket instance)
• Receive queue owns buffered message data
• recv() allocates message buffer, caller must free

Author

Zachary Fogg me@zf.nosp@m.o.gg

Date

February 2026

Definition in file websocket/transport.c.

Macro Definition Documentation

WEBSOCKET_MESSAGE_QUEUE_SIZE_INCOMING

#define WEBSOCKET_MESSAGE_QUEUE_SIZE_INCOMING   64

Maximum incoming message queue size (frames received from peer)

Set to 64 for balance: allows buffering across one large frame's fragments while keeping latency reasonable. With ~70 fragments per 291KB frame, we need at least 64 slots to avoid dropping the frame while it's fragmenting.

Definition at line 60 of file websocket/transport.c.

WEBSOCKET_MESSAGE_QUEUE_SIZE_OUTGOING

#define WEBSOCKET_MESSAGE_QUEUE_SIZE_OUTGOING   512

Maximum outgoing message queue size (frames to send to peer)

Set to 64 to match incoming queue and prevent "BINARY after FIN" errors. Large video frames fragment into ~70 chunks; queue must be large enough to buffer one complete message worth of fragments.

Definition at line 69 of file websocket/transport.c.

WEBSOCKET_RECV_QUEUE_SIZE

#define WEBSOCKET_RECV_QUEUE_SIZE   WEBSOCKET_MESSAGE_QUEUE_SIZE_INCOMING

Definition at line 72 of file websocket/transport.c.

WEBSOCKET_SEND_QUEUE_SIZE

#define WEBSOCKET_SEND_QUEUE_SIZE   WEBSOCKET_MESSAGE_QUEUE_SIZE_OUTGOING

Definition at line 73 of file websocket/transport.c.

Function Documentation

acip_websocket_client_transport_create()

acip_transport_t * acip_websocket_client_transport_create	(	const char *	name,
		const char *	url,
		crypto_context_t *	crypto_ctx
	)

Create WebSocket client transport.

Parameters

url	WebSocket URL (e.g., "ws://localhost:27225")
crypto_ctx	Optional encryption context (can be NULL)

Returns

Transport instance or NULL on failure

Definition at line 1412 of file websocket/transport.c.

                                                                                       {
  if (!name) {
    SET_ERRNO(ERROR_INVALID_STATE, "Transport name is required");
    return NULL;
  }
 
  if (!url) {
    SET_ERRNO(ERROR_INVALID_PARAM, "url is required");
    return NULL;
  }
 
  // Parse URL to extract host, port, and path
  // Format: ws://host:port/path or wss://host:port/path
  const char *protocol_end = strstr(url, "://");
  if (!protocol_end) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid WebSocket URL format (missing ://)");
    return NULL;
  }
 
  bool use_ssl = (strncmp(url, "wss://", 6) == 0);
  const char *host_start = protocol_end + 3;
 
  // Find port (if specified)
  const char *port_start = strchr(host_start, ':');
  const char *path_start = strchr(host_start, '/');
 
  char host[256] = {0};
  int port = use_ssl ? 443 : 27226; // Default: wss:// uses 443, ws:// uses 27226 (ascii-chat WebSocket port)
  char path[256] = "/";
 
  if (port_start && (!path_start || port_start < path_start)) {
    // Port is specified
    size_t host_len = port_start - host_start;
    if (host_len >= sizeof(host)) {
      SET_ERRNO(ERROR_INVALID_PARAM, "Host name too long");
      return NULL;
    }
    memcpy(host, host_start, host_len);
    host[host_len] = '\0';
 
    // Extract port
    char *endptr;
    errno = 0;
    long port_val = strtol(port_start + 1, &endptr, 10);
    if (*endptr != '\0' || errno != 0 || port_val <= 0 || port_val > 65535) {
      log_error("websocket_transport_create: Invalid port number: %s", port_start + 1);
      SET_ERRNO(ERROR_INVALID_PARAM, "Invalid port number");
      return NULL;
    }
    port = (uint16_t)port_val;
  } else {
    // No port specified, use default
    size_t host_len = path_start ? (size_t)(path_start - host_start) : strlen(host_start);
    if (host_len >= sizeof(host)) {
      SET_ERRNO(ERROR_INVALID_PARAM, "Host name too long");
      return NULL;
    }
    memcpy(host, host_start, host_len);
    host[host_len] = '\0';
  }
 
  // Extract path
  if (path_start) {
    SAFE_STRNCPY(path, path_start, sizeof(path) - 1);
    path[sizeof(path) - 1] = '\0';
  }
 
  log_info("Connecting to WebSocket: %s (host=%s, port=%d, path=%s, ssl=%d)", url, host, port, path, use_ssl);
 
  // Allocate transport structure
  acip_transport_t *transport = SAFE_MALLOC(sizeof(acip_transport_t), acip_transport_t *);
  if (!transport) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate WebSocket transport");
    return NULL;
  }
 
  // Allocate WebSocket-specific data
  websocket_transport_data_t *ws_data =
      SAFE_CALLOC(1, sizeof(websocket_transport_data_t), websocket_transport_data_t *);
  if (!ws_data) {
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate WebSocket transport data");
    return NULL;
  }
 
  // Initialize destruction state flag (CALLOC zero-initializes, but explicit for clarity)
  atomic_store_bool(&ws_data->is_destroying, false);
 
  // Create receive queue
  ws_data->recv_queue = ringbuffer_create(sizeof(websocket_recv_msg_t), WEBSOCKET_RECV_QUEUE_SIZE);
  if (!ws_data->recv_queue) {
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create receive queue");
    return NULL;
  }
 
  // Create send queue for client transport to buffer outgoing messages
  ws_data->send_queue = ringbuffer_create(sizeof(websocket_msg_t), WEBSOCKET_SEND_QUEUE_SIZE);
  if (!ws_data->send_queue) {
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create send queue");
    return NULL;
  }
 
  // Initialize synchronization primitives with transport-aware names
  char recv_name[64], state_name[64];
  snprintf(recv_name, sizeof(recv_name), "recv_%s", name);
  snprintf(state_name, sizeof(state_name), "state_%s", name);
 
  if (mutex_init(&ws_data->recv_mutex, recv_name) != 0) {
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize recv mutex");
    return NULL;
  }
 
  if (cond_init(&ws_data->recv_cond, recv_name) != 0) {
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize recv condition variable");
    return NULL;
  }
 
  if (mutex_init(&ws_data->state_mutex, state_name) != 0) {
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize state mutex");
    return NULL;
  }
 
  if (cond_init(&ws_data->state_cond, state_name) != 0) {
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize state condition variable");
    return NULL;
  }
 
  // Allocate initial send buffer
  ws_data->send_buffer_capacity = LWS_PRE + 524288; // Match tx_packet_size (512KB)
  ws_data->send_buffer = SAFE_MALLOC(ws_data->send_buffer_capacity, uint8_t *);
  if (!ws_data->send_buffer) {
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate send buffer");
    return NULL;
  }
 
  // Create libwebsockets context
  // Protocol array must persist for lifetime of context - use static
  // Register "http" protocol for WebSocket connections - matches server-side protocol name
  // This ensures libwebsockets will invoke our websocket_callback during upgrade handshake and communication
  static struct lws_protocols client_protocols[] = {
      {
          "http", // Default HTTP protocol - handles WebSocket upgrade and ACIP communication
          websocket_callback,
          0,      // Per-session data - using connect_info.userdata instead
          524288, // RX buffer size
          0,      // ID
          NULL,   // User pointer (set from connect_info.userdata)
          524288  // TX packet size
      },
      {
          "acip", // Application subprotocol selected by the ACIP WebSocket client
          websocket_callback,
          0,
          524288,
          0,
          NULL,
          524288
      },
      {NULL, NULL, 0, 0, 0, NULL, 0} // Terminator
  };
 
  // Disable client compression for now - causes assertion in lws_set_extension_option()
  // This is a known issue with libwebsockets permessage-deflate negotiation
  // Server-side compression is still enabled, data will be compressed from server to client
  // but client->server traffic remains uncompressed (acceptable since client sends less data)
  // static const struct lws_extension client_extensions[] = {
  //     {"permessage-deflate", lws_extension_callback_pm_deflate, "permessage-deflate; client_max_window_bits=15"},
  //     {NULL, NULL, NULL}};
 
  struct lws_context_creation_info info;
  memset(&info, 0, sizeof(info));
  info.port = CONTEXT_PORT_NO_LISTEN; // Client mode - no listening
  info.protocols = client_protocols;
  info.gid = (gid_t)-1; // Cast to avoid undefined behavior with unsigned type
  info.uid = (uid_t)-1; // Cast to avoid undefined behavior with unsigned type
  // Initialize SSL globally for client (required even for clients to use OpenSSL)
  // The LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT flag works for both servers and clients
  info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
  info.extensions = NULL; // Disable client compression due to lws_set_extension_option() assertion
 
  // Increase per-thread service buffer to prevent fragmentation of large messages
  // Default is 4KB, causing 291KB frames to fragment into 73 × 4KB chunks
  // Increase to 512KB to match server and allow larger WebSocket frames without fragmentation
  info.pt_serv_buf_size = 512 * 1024; // 512KB per-thread service buffer
 
  // Configure keep-alive to prevent idle disconnects during handshake
  // PING every 30 seconds if idle, close after 35 seconds total with no response
  static const lws_retry_bo_t client_keep_alive_policy = {
      .secs_since_valid_ping = 30,   // Send PING after 30s idle
      .secs_since_valid_hangup = 35, // Hangup if still idle after 35s
  };
  info.retry_and_idle_policy = &client_keep_alive_policy;
 
  // Enable libwebsockets logging through centralized logging system
  lws_log_init_client();
 
  ws_data->context = lws_create_context(&info);
 
  if (!ws_data->context) {
    SAFE_FREE(ws_data->send_buffer);
    cond_destroy(&ws_data->state_cond);
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to create libwebsockets context");
    return NULL;
  }
 
  // Initialize connection state flags BEFORE calling lws_client_connect_via_info
  // The callback might fire immediately, so all fields must be ready
  ws_data->is_connected = false;      // Will be set to true in LWS_CALLBACK_CLIENT_ESTABLISHED
  ws_data->connection_failed = false; // Set to true in LWS_CALLBACK_CLIENT_CONNECTION_ERROR
  ws_data->owns_context = true;       // Client transport owns the context
 
  // Connect to WebSocket server
  log_debug("Initiating WebSocket connection to %s:%d%s", host, port, path);
  struct lws_client_connect_info connect_info;
  memset(&connect_info, 0, sizeof(connect_info));
  connect_info.context = ws_data->context;
  connect_info.address = host;
  connect_info.port = port;
  connect_info.path = path;
  connect_info.host = host;
  connect_info.origin = host;
  // Request the ACIP subprotocol so server routes to acip_protocol handler
  // This ensures WebSocket upgrade includes Sec-WebSocket-Protocol: acip
  connect_info.local_protocol_name = "acip";
  connect_info.protocol = "acip";
  // Use SSL + skip server certificate hostname verification + allow self-signed certs (for development)
  connect_info.ssl_connection =
      use_ssl ? (LCCSCF_USE_SSL | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK | LCCSCF_ALLOW_SELFSIGNED) : 0;
  connect_info.opaque_user_data = ws_data;
 
  log_debug("Calling lws_client_connect_via_info...");
  ws_data->wsi = lws_client_connect_via_info(&connect_info);
  log_debug("lws_client_connect_via_info returned: %p", (void *)ws_data->wsi);
  if (!ws_data->wsi) {
    lws_context_destroy(ws_data->context);
    SAFE_FREE(ws_data->send_buffer);
    cond_destroy(&ws_data->state_cond);
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to connect to WebSocket server");
    return NULL;
  }
 
  // Initialize transport
  transport->methods = &websocket_methods;
  transport->crypto_ctx = crypto_ctx;
  transport->impl_data = ws_data;
 
  // Start service thread immediately - it will drive TLS handshake with frequent polling
  // The service thread checks is_destroying and context validity before calling lws_service,
  // so it's safe to start it right away without delays.
  ws_data->service_running = true;
  if (asciichat_thread_create(&ws_data->service_thread, "ws_service", websocket_service_thread, ws_data) != 0) {
    log_error("Failed to create WebSocket service thread");
    ws_data->service_running = false;
    lws_context_destroy(ws_data->context);
    SAFE_FREE(ws_data->send_buffer);
    cond_destroy(&ws_data->state_cond);
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to create service thread");
    return NULL;
  }
  log_debug("WebSocket service thread started");
 
  // CRITICAL FIX: Do NOT block the main thread waiting for connection!
  // This prevents stdin/keyboard input from being processed and causes the client to hang.
  // The service thread will establish the connection asynchronously.
  // The protocol layer (recv) will detect if connection fails and handle it there.
  // Return immediately so the main thread can respond to keyboard input.
  log_debug("WebSocket transport created, service thread will establish connection asynchronously");
 
  // Register transport
  NAMED_REGISTER_TRANSPORT(transport, name, NULL);
 
  // Register websocket implementation data with transport as parent
  (void)NAMED_REGISTER_WEBSOCKET_IMPL(ws_data, "impl", (uintptr_t)(const void *)(transport));
 
  // Return transport immediately - connection will be established by service thread
  // If connection fails, recv() will detect is_connected=false and return error
  return transport;
}

References asciichat_thread_create, atomic_store_bool(), cond_destroy(), cond_init(), websocket_transport_data_t::connection_failed, websocket_transport_data_t::context, acip_transport::crypto_ctx, errno, ERROR_INTERNAL, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, ERROR_NETWORK, acip_transport::impl_data, websocket_transport_data_t::is_connected, websocket_transport_data_t::is_destroying, log_debug, log_error, log_info, lws_log_init_client(), acip_transport::methods, mutex_destroy(), mutex_init(), NAMED_REGISTER_TRANSPORT, NAMED_REGISTER_WEBSOCKET_IMPL, websocket_transport_data_t::owns_context, websocket_transport_data_t::recv_cond, websocket_transport_data_t::recv_mutex, websocket_transport_data_t::recv_queue, ringbuffer_create(), ringbuffer_destroy(), SAFE_CALLOC, SAFE_FREE, SAFE_MALLOC, SAFE_STRNCPY, websocket_transport_data_t::send_buffer, websocket_transport_data_t::send_buffer_capacity, websocket_transport_data_t::send_queue, websocket_transport_data_t::service_running, websocket_transport_data_t::service_thread, SET_ERRNO, websocket_transport_data_t::state_cond, websocket_transport_data_t::state_mutex, WEBSOCKET_RECV_QUEUE_SIZE, WEBSOCKET_SEND_QUEUE_SIZE, and websocket_transport_data_t::wsi.

Referenced by connection_factory_open(), and websocket_client_connect().

acip_websocket_server_transport_create()

acip_transport_t * acip_websocket_server_transport_create	(	const char *	name,
		struct lws *	wsi,
		crypto_context_t *	crypto_ctx
	)

Create WebSocket server transport from existing connection.

Wraps an already-established libwebsockets connection (from server accept). Used by websocket_server module to create transports for incoming clients.

Parameters

wsi	Established libwebsockets connection (not owned by transport)
crypto_ctx	Optional crypto context

Returns

Transport instance or NULL on error

Definition at line 1757 of file websocket/transport.c.

                                                                                       {
  if (!name) {
    SET_ERRNO(ERROR_INVALID_STATE, "Transport name is required");
    return NULL;
  }
 
  if (!wsi) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid wsi parameter");
    return NULL;
  }
 
  // Allocate transport structure
  acip_transport_t *transport = SAFE_CALLOC(1, sizeof(acip_transport_t), acip_transport_t *);
  if (!transport) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate WebSocket transport");
    return NULL;
  }
 
  // Allocate transport-specific data
  websocket_transport_data_t *ws_data =
      SAFE_CALLOC(1, sizeof(websocket_transport_data_t), websocket_transport_data_t *);
  if (!ws_data) {
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate WebSocket transport data");
    return NULL;
  }
 
  // Initialize receive queue
  ws_data->recv_queue = ringbuffer_create(sizeof(websocket_recv_msg_t), WEBSOCKET_RECV_QUEUE_SIZE);
  if (!ws_data->recv_queue) {
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create receive queue");
    return NULL;
  }
 
  // Initialize send queue (for server-side transports)
  ws_data->send_queue = ringbuffer_create(sizeof(websocket_recv_msg_t), WEBSOCKET_SEND_QUEUE_SIZE);
  if (!ws_data->send_queue) {
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create send queue");
    return NULL;
  }
 
  // Initialize pending-free queue for deferred buffer freeing
  // permessage-deflate compression holds buffer references asynchronously,
  // so we defer freeing to prevent use-after-free errors
  ws_data->pending_free_queue = ringbuffer_create(sizeof(pending_free_item_t), 256);
  if (!ws_data->pending_free_queue) {
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create pending-free queue");
    return NULL;
  }
 
  // Initialize synchronization primitives with transport-aware names
  char recv_name[64], send_name[64], state_name[64], pending_free_name[64];
  snprintf(recv_name, sizeof(recv_name), "recv_%s", name);
  snprintf(send_name, sizeof(send_name), "send_%s", name);
  snprintf(state_name, sizeof(state_name), "state_%s", name);
  snprintf(pending_free_name, sizeof(pending_free_name), "pending_free_%s", name);
 
  if (mutex_init(&ws_data->recv_mutex, recv_name) != 0) {
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize recv mutex");
    return NULL;
  }
 
  if (cond_init(&ws_data->recv_cond, recv_name) != 0) {
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize recv condition variable");
    return NULL;
  }
 
  if (mutex_init(&ws_data->send_mutex, send_name) != 0) {
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize send mutex");
    return NULL;
  }
 
  if (mutex_init(&ws_data->state_mutex, state_name) != 0) {
    mutex_destroy(&ws_data->send_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize state mutex");
    return NULL;
  }
 
  if (cond_init(&ws_data->state_cond, state_name) != 0) {
    mutex_destroy(&ws_data->state_mutex);
    mutex_destroy(&ws_data->send_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize state condition variable");
    return NULL;
  }
 
  if (mutex_init(&ws_data->pending_free_mutex, pending_free_name) != 0) {
    cond_destroy(&ws_data->state_cond);
    mutex_destroy(&ws_data->state_mutex);
    mutex_destroy(&ws_data->send_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    ringbuffer_destroy(ws_data->pending_free_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to initialize pending-free mutex");
    return NULL;
  }
 
  // Allocate send buffer with LWS_PRE padding
  size_t initial_capacity = 4096 + LWS_PRE;
  ws_data->send_buffer = SAFE_MALLOC(initial_capacity, uint8_t *);
  if (!ws_data->send_buffer) {
    mutex_destroy(&ws_data->state_mutex);
    mutex_destroy(&ws_data->send_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    ringbuffer_destroy(ws_data->send_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate send buffer");
    return NULL;
  }
  ws_data->send_buffer_capacity = initial_capacity;
 
  // Store connection info (server-side: no context ownership, connection already established)
  ws_data->wsi = wsi;
  ws_data->context = lws_get_context(wsi); // Get context from wsi (not owned)
  ws_data->owns_context = false;           // Server owns context, not transport
  ws_data->is_connected = true;            // Already connected (server-side)
  log_info("[WEBSOCKET_TRANSPORT_CREATE] ★★★ SERVER TRANSPORT CREATED: is_connected=true, wsi=%p, ws_data=%p",
           (void *)wsi, (void *)ws_data);
  log_debug("Server transport created: is_connected=true, wsi=%p", (void *)wsi);
 
  // Initialize transport
  transport->methods = &websocket_methods;
  transport->crypto_ctx = crypto_ctx;
  transport->impl_data = ws_data;
 
  log_info("Created WebSocket server transport (crypto: %s)", crypto_ctx ? "enabled" : "disabled");
 
  // Register transport
  NAMED_REGISTER_TRANSPORT(transport, name, NULL);
 
  // Register websocket implementation data with transport as parent
  (void)NAMED_REGISTER_WEBSOCKET_IMPL(ws_data, "impl", (uintptr_t)(const void *)(transport));
 
  return transport;
}

References cond_destroy(), cond_init(), websocket_transport_data_t::context, acip_transport::crypto_ctx, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, ERROR_NETWORK, acip_transport::impl_data, websocket_transport_data_t::is_connected, log_debug, log_info, acip_transport::methods, mutex_destroy(), mutex_init(), NAMED_REGISTER_TRANSPORT, NAMED_REGISTER_WEBSOCKET_IMPL, websocket_transport_data_t::owns_context, websocket_transport_data_t::pending_free_mutex, websocket_transport_data_t::pending_free_queue, websocket_transport_data_t::recv_cond, websocket_transport_data_t::recv_mutex, websocket_transport_data_t::recv_queue, ringbuffer_create(), ringbuffer_destroy(), SAFE_CALLOC, SAFE_FREE, SAFE_MALLOC, websocket_transport_data_t::send_buffer, websocket_transport_data_t::send_buffer_capacity, websocket_transport_data_t::send_mutex, websocket_transport_data_t::send_queue, SET_ERRNO, websocket_transport_data_t::state_cond, websocket_transport_data_t::state_mutex, WEBSOCKET_RECV_QUEUE_SIZE, WEBSOCKET_SEND_QUEUE_SIZE, and websocket_transport_data_t::wsi.