WebSocket transport implementation for ACIP protocol. More...

Macros
#define	WEBSOCKET_RECV_QUEUE_SIZE 4096
	Maximum receive queue size (messages buffered before recv())

#define	WEBSOCKET_SEND_QUEUE_SIZE 256
	Maximum send queue size (messages buffered for server-side sending)

Functions
acip_transport_t *	acip_websocket_client_transport_create (const char url, crypto_context_t crypto_ctx)
	Create WebSocket client transport.

acip_transport_t *	acip_websocket_server_transport_create (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:

send(): Synchronous write via lws_write()
LWS callback: Async write to receive ringbuffer
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_RECV_QUEUE_SIZE

#define WEBSOCKET_RECV_QUEUE_SIZE 4096

Maximum receive queue size (messages buffered before recv())

Power of 2 for ringbuffer optimization. Increased from 512 to buffer multiple large frames and reduce queue pressure. Each slot holds one message (up to 921KB). With 4096 slots, can buffer ~3.7GB.

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

◆ WEBSOCKET_SEND_QUEUE_SIZE

#define WEBSOCKET_SEND_QUEUE_SIZE 256

Maximum send queue size (messages buffered for server-side sending)

Larger than receive queue because video frames are continuously sent. Must be large enough to buffer frames while event loop processes them.

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

Function Documentation

◆ acip_websocket_client_transport_create()

acip_transport_t * acip_websocket_client_transport_create	(	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 702 of file websocket/transport.c.

                                                                                                        {
  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
    port = atoi(port_start + 1);
    if (port <= 0 || port > 65535) {
      SET_ERRNO(ERROR_INVALID_PARAM, "Invalid port number");
      return NULL;
    }
  } 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;
  }
 
  // 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;
  }
 
  // Initialize synchronization primitives
  if (mutex_init(&ws_data->recv_mutex) != 0) {
    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) != 0) {
    mutex_destroy(&ws_data->recv_mutex);
    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) != 0) {
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    ringbuffer_destroy(ws_data->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize state mutex");
    return NULL;
  }
 
  // Allocate initial send buffer
  ws_data->send_buffer_capacity = LWS_PRE + 8192; // Initial 8KB buffer
  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->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
  static struct lws_protocols client_protocols[] = {
      {"acip", websocket_callback, 0, 4096, 0, NULL, 0}, {NULL, NULL, 0, 0, 0, NULL, 0} // Terminator
  };
 
  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
  info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
 
  ws_data->context = lws_create_context(&info);
  if (!ws_data->context) {
    SAFE_FREE(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->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "Failed to create libwebsockets context");
    return NULL;
  }
 
  // 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;
  connect_info.protocol = "acip";
  connect_info.ssl_connection = use_ssl ? LCCSCF_USE_SSL : 0;
  connect_info.userdata = 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);
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    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;
  }
 
  ws_data->is_connected = false; // Will be set to true in LWS_CALLBACK_CLIENT_ESTABLISHED
  ws_data->owns_context = true;  // Client transport owns the context
 
  // Initialize transport
  transport->methods = &websocket_methods;
  transport->crypto_ctx = crypto_ctx;
  transport->impl_data = ws_data;
 
  // Wait for connection to establish (synchronous connection)
  // Service the libwebsockets event loop until connected or timeout
  log_debug("Waiting for WebSocket connection to establish...");
  int timeout_ms = 5000; // 5 second timeout
  int elapsed_ms = 0;
  while (!ws_data->is_connected && elapsed_ms < timeout_ms) {
    // Service libwebsockets (processes network events, triggers callbacks)
    int result = lws_service(ws_data->context, 50); // 50ms timeout per iteration
    if (result < 0) {
      log_error("lws_service error during connection: %d", result);
      lws_context_destroy(ws_data->context);
      SAFE_FREE(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->recv_queue);
      SAFE_FREE(ws_data);
      SAFE_FREE(transport);
      SET_ERRNO(ERROR_NETWORK, "WebSocket connection failed");
      return NULL;
    }
    elapsed_ms += 50;
  }
 
  if (!ws_data->is_connected) {
    log_error("WebSocket connection timeout after %d ms", elapsed_ms);
    lws_context_destroy(ws_data->context);
    SAFE_FREE(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->recv_queue);
    SAFE_FREE(ws_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_NETWORK, "WebSocket connection timeout");
    return NULL;
  }
 
  log_info("WebSocket connection established (crypto: %s)", crypto_ctx ? "enabled" : "disabled");
 
  // Start service thread to process incoming messages
  ws_data->service_running = true;
  if (asciichat_thread_create(&ws_data->service_thread, 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);
    mutex_destroy(&ws_data->state_mutex);
    cond_destroy(&ws_data->recv_cond);
    mutex_destroy(&ws_data->recv_mutex);
    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 for client transport");
 
  return transport;
}

References asciichat_thread_create(), mutex_destroy(), mutex_init(), ringbuffer_create(), ringbuffer_destroy(), and WEBSOCKET_RECV_QUEUE_SIZE.

Referenced by server_connection_establish(), and websocket_client_connect().

◆ acip_websocket_server_transport_create()

acip_transport_t * acip_websocket_server_transport_create	(	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.