transport.h File Reference

Transport abstraction layer for ACIP protocol. More...

Go to the source code of this file.

Data Structures
struct	acip_transport_methods_t
	Transport method table (virtual function table) More...
struct	acip_transport
	Transport instance structure. More...

Typedefs
typedef struct acip_transport	acip_transport_t
	Forward declaration of transport implementation.

Enumerations
enum	acip_transport_type_t {   ACIP_TRANSPORT_TCP = 1 , ACIP_TRANSPORT_WEBSOCKET = 2 , ACIP_TRANSPORT_WEBRTC = 3 , ACIP_TRANSPORT_HTTP = 4 ,   ACIP_TRANSPORT_QUIC = 5 }
	Transport type enumeration. More...

Functions
void	acip_transport_destroy (acip_transport_t *transport)
	Destroy transport and free all resources.
acip_transport_t *	acip_tcp_transport_create (const char *name, socket_t sockfd, crypto_context_t *crypto_ctx)
	Create TCP transport from existing socket.
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.
acip_transport_t *	acip_webrtc_transport_create (struct webrtc_peer_connection *peer_conn, struct webrtc_data_channel *data_channel, crypto_context_t *crypto_ctx)

Detailed Description

Transport abstraction layer for ACIP protocol.

Provides a transport-agnostic interface that allows ACIP protocol handlers to work with any underlying transport (TCP, WebSocket, HTTP, etc.).

Transport Layer Design

Architecture

The transport layer abstracts protocol delivery, allowing the same ACIP protocol handlers to work with multiple underlying transports:

• TCP (acip_tcp_transport_create): Raw TCP sockets, binary packet streaming
• WebSocket (acip_websocket_client_transport_create): HTTP upgrade, frame-based
• WebRTC (acip_webrtc_transport_create): P2P data channels, encryption at transport
• HTTP (future): Long-polling, REST-style
• QUIC (future): UDP with congestion control

Design Principles

1. Protocol code never calls socket()/send()/recv() directly
2. Same ACIP handlers work with any transport implementation
3. Each transport handles its own connection state and reliability
4. Clean separation between protocol logic (acip_*) and transport logic

Packet Framing

Transports handle packet framing transparently:

• TCP: Uses packet_header_t (magic + type + length + CRC32 + client_id)
• WebSocket: Wraps packets in WebSocket frames
• WebRTC: Sends packets directly on data channel

Version Information

• Transport API Version: 1.0 (stable)
• Supported ACIP Version: 1.0
• Release Date: January 2026

DESIGN GOALS:

1. Protocol code never calls socket()/send()/recv() directly
2. Same ACIP handlers work with TCP, WebSocket, or any future transport
3. Transport implementations handle their own connection state
4. Clean separation between protocol logic and transport logic

USAGE PATTERN:

// Create TCP transport from existing socket
acip_transport_t *tcp = acip_tcp_transport_create(sockfd, crypto_ctx);
 
// Send packet with type-specific helper
asciichat_error_t result = acip_send_ascii_frame(tcp, frame_data, frame_size, ...);
 
// Receive packet with dispatcher
acip_client_receive_and_dispatch(tcp, &callbacks);
 
// Or create WebSocket client transport
acip_transport_t *ws = acip_websocket_client_transport_create("ws://localhost:27225", crypto_ctx);
asciichat_error_t result = acip_send_ascii_frame(ws, frame_data, frame_size, ...);
 
// Cleanup
acip_transport_destroy(tcp);
acip_transport_destroy(ws);

MEMORY OWNERSHIP:

• Transport owns its connection state (socket, pointers, etc.)
• send() does NOT take ownership of data (caller must keep it valid)
• recv() allocates buffer, caller must free via out_allocated_buffer
• destroy() cleans up all transport resources and frees transport structure

Author

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

Date

January 2026

Version

1.0 (Transport API)

Definition in file transport.h.

Typedef Documentation

acip_transport_t

typedef struct acip_transport acip_transport_t

Forward declaration of transport implementation.

Opaque handle to transport-specific state. Each transport implementation defines its own state structure.

Definition at line 107 of file transport.h.

Enumeration Type Documentation

acip_transport_type_t

enum acip_transport_type_t

Transport type enumeration.

Identifies which transport implementation is being used. Useful for debugging and conditional logic.

Enumerator
ACIP_TRANSPORT_TCP	Raw TCP socket.
ACIP_TRANSPORT_WEBSOCKET	WebSocket over TCP.
ACIP_TRANSPORT_WEBRTC	WebRTC DataChannel (P2P)
ACIP_TRANSPORT_HTTP	HTTP long-polling (future)
ACIP_TRANSPORT_QUIC	QUIC/UDP (future)

Definition at line 93 of file transport.h.

             {
  ACIP_TRANSPORT_TCP = 1,       
  ACIP_TRANSPORT_WEBSOCKET = 2, 
  ACIP_TRANSPORT_WEBRTC = 3,    
  ACIP_TRANSPORT_HTTP = 4,      
  ACIP_TRANSPORT_QUIC = 5       
} acip_transport_type_t;

Function Documentation

acip_tcp_transport_create()

acip_transport_t * acip_tcp_transport_create	(	const char *	name,
		socket_t	sockfd,
		crypto_context_t *	crypto_ctx
	)

Create TCP transport from existing socket.

Parameters

name	Debug name for tracking this transport (required, e.g., "transport_client_5")
sockfd	Connected socket descriptor
crypto_ctx	Optional crypto context (may be NULL)

Returns

Transport instance or NULL on error

Note

Caller retains socket ownership (transport doesn't close on destroy)

Definition at line 398 of file tcp/transport.c.

                                                                                                             {
  if (!name) {
    SET_ERRNO(ERROR_INVALID_STATE, "Transport name is required");
    return NULL;
  }
 
  if (sockfd == INVALID_SOCKET_VALUE) {
    log_error("[TCP_CREATE_STATE] ❌ INVALID_SOCKET: sockfd=%d", sockfd);
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid socket descriptor");
    return NULL;
  }
 
  log_info("[TCP_CREATE_STATE] 🟢 CREATE_START: sockfd=%d, crypto=%s", sockfd, crypto_ctx ? "yes" : "no");
 
  // Allocate transport structure
  acip_transport_t *transport = SAFE_MALLOC(sizeof(acip_transport_t), acip_transport_t *);
  if (!transport) {
    log_error("[TCP_CREATE_STATE] ❌ ALLOC_TRANSPORT_FAILED: size=%zu", sizeof(acip_transport_t));
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate TCP transport");
    return NULL;
  }
 
  log_debug("[TCP_CREATE_STATE] 🟢 TRANSPORT_ALLOCATED: transport=%p", (void *)transport);
 
  // Allocate TCP-specific data
  tcp_transport_data_t *tcp_data = SAFE_MALLOC(sizeof(tcp_transport_data_t), tcp_transport_data_t *);
  if (!tcp_data) {
    log_error("[TCP_CREATE_STATE] ❌ ALLOC_TCP_DATA_FAILED: size=%zu", sizeof(tcp_transport_data_t));
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate TCP transport data");
    return NULL;
  }
 
  log_debug("[TCP_CREATE_STATE] 🟢 TCP_DATA_ALLOCATED: tcp_data=%p", (void *)tcp_data);
 
  // Initialize TCP data
  tcp_data->sockfd = sockfd;
  tcp_data->is_connected = true;
 
  // Initialize send mutex to protect concurrent sends from multiple threads
  int mutex_result = mutex_init(&tcp_data->send_mutex, "tcp_send");
  if (mutex_result != 0) {
    log_error("[TCP_CREATE_STATE] ❌ MUTEX_INIT_FAILED: result=%d", mutex_result);
    SAFE_FREE(tcp_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to initialize send mutex");
    return NULL;
  }
 
  log_info("[TCP_CREATE_STATE] 🟢 TCP_DATA_INITIALIZED: sockfd=%d, is_connected=true, send_mutex=initialized", sockfd);
 
  // Enable TCP_NODELAY to disable Nagle's algorithm
  // This ensures small packets are sent immediately instead of being buffered
  int nodelay = 1;
  if (setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, (const char *)&nodelay, sizeof(nodelay)) < 0) {
    log_warn("[TCP_CREATE_STATE] ⚠️  TCP_NODELAY_FAILED: sockfd=%d, %s", sockfd, SAFE_STRERROR(errno));
    // Continue anyway - this is not fatal
  } else {
    log_debug("[TCP_CREATE_STATE] 🟢 TCP_NODELAY_SET: sockfd=%d", sockfd);
  }
 
  // Initialize transport
  transport->methods = &tcp_methods;
  transport->crypto_ctx = crypto_ctx;
  transport->impl_data = tcp_data;
 
  // Register transport and impl_data before logging so named replacements work
  NAMED_REGISTER_TRANSPORT(transport, name, NULL);
  NAMED_REGISTER(tcp_data, "impl", "tcp_impl", "0x%tx", (uintptr_t)(const void *)(transport));
 
  log_info("[TCP_CREATE_STATE] ✅ CREATE_COMPLETE: transport=%p, sockfd=%d, is_connected=true, crypto=%s",
           (void *)transport, sockfd, crypto_ctx ? "enabled" : "disabled");
 
  return transport;
}

References errno, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, INVALID_SOCKET_VALUE, tcp_transport_data_t::is_connected, log_debug, log_error, log_info, log_warn, mutex_init(), NAMED_REGISTER, NAMED_REGISTER_TRANSPORT, SAFE_FREE, SAFE_MALLOC, SAFE_STRERROR, tcp_transport_data_t::send_mutex, SET_ERRNO, and tcp_transport_data_t::sockfd.

Referenced by acds_client_connect(), acds_client_handler(), add_client(), connection_attempt_tcp(), and server_connection_establish().

acip_transport_destroy()

void acip_transport_destroy

(

acip_transport_t *

transport

)

Destroy transport and free all resources.

Parameters

transport

Transport to destroy (may be NULL)

Note

Calls close() first if still connected

Frees the transport structure itself

Safe to pass NULL

Definition at line 478 of file tcp/transport.c.

                                                         {
  if (!transport) {
    log_debug("[TRANSPORT_DESTROY] ⚠️  NULL_TRANSPORT: nothing to destroy");
    return;
  }
 
  log_warn("[TRANSPORT_DESTROY] 🔴 DESTROY_START: transport=%p, impl_data=%p", (void *)transport, transport->impl_data);
 
  // Get type before we destroy for logging
  acip_transport_type_t type = 0;
  if (transport->methods && transport->methods->get_type) {
    type = transport->methods->get_type(transport);
    log_info("[TRANSPORT_DESTROY] 📋 TRANSPORT_TYPE: type=%d (1=TCP, 2=WebSocket, 3=WebRTC, ...)", type);
  }
 
  // Close if still connected
  if (transport->methods && transport->methods->close && transport->methods->is_connected &&
      transport->methods->is_connected(transport)) {
    log_warn("[TRANSPORT_DESTROY] 🔴 STILL_CONNECTED: calling close() first (type=%d)", type);
    asciichat_error_t close_result = transport->methods->close(transport);
    log_info("[TRANSPORT_DESTROY] 🔴 CLOSE_CALLED: result=%d", close_result != ASCIICHAT_OK ? -1 : 0);
  } else {
    log_debug("[TRANSPORT_DESTROY] ✅ ALREADY_CLOSED: skipping close (type=%d)", type);
  }
 
  // Call custom destroy implementation if provided
  if (transport->methods && transport->methods->destroy_impl) {
    log_info("[TRANSPORT_DESTROY] 🔧 CALLING_DESTROY_IMPL: type=%d", type);
    transport->methods->destroy_impl(transport);
    log_info("[TRANSPORT_DESTROY] ✅ DESTROY_IMPL_COMPLETE: type=%d", type);
  } else {
    log_debug("[TRANSPORT_DESTROY] ⏭️  NO_CUSTOM_DESTROY: type=%d (using generic cleanup)", type);
  }
 
  // Free implementation data
  if (transport->impl_data) {
    log_debug("[TRANSPORT_DESTROY] 🗑️  FREEING_IMPL_DATA: %p", transport->impl_data);
    NAMED_UNREGISTER(transport->impl_data);
    SAFE_FREE(transport->impl_data);
    log_debug("[TRANSPORT_DESTROY] ✅ IMPL_DATA_FREED");
  }
 
  // Free transport structure
  log_debug("[TRANSPORT_DESTROY] 🗑️  FREEING_TRANSPORT: %p", (void *)transport);
  SAFE_FREE(transport);
 
  log_warn("[TRANSPORT_DESTROY] ✅ DESTROY_COMPLETE: type=%d", type);
}

References ASCIICHAT_OK, acip_transport_methods_t::close, acip_transport_methods_t::destroy_impl, acip_transport_methods_t::get_type, acip_transport::impl_data, acip_transport_methods_t::is_connected, log_debug, log_info, log_warn, acip_transport::methods, NAMED_UNREGISTER, and SAFE_FREE.

Referenced by acds_client_connect(), acds_client_disconnect(), acds_client_handler(), connection_attempt_tcp(), connection_handle_cleanup(), discovery_session_destroy(), discovery_session_process(), remove_client(), server_connection_close(), server_connection_establish(), server_connection_set_transport(), and session_participant_destroy().

acip_webrtc_transport_create()

acip_transport_t * acip_webrtc_transport_create	(	struct webrtc_peer_connection *	peer_conn,
		struct webrtc_data_channel *	data_channel,
		crypto_context_t *	crypto_ctx
	)

Definition at line 614 of file webrtc/transport.c.

                                                                             {
  if (!peer_conn || !data_channel) {
    SET_ERRNO(ERROR_INVALID_PARAM, "peer_conn and data_channel are required");
    return NULL;
  }
 
  // 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 WebRTC transport");
    return NULL;
  }
 
  // Allocate WebRTC-specific data
  webrtc_transport_data_t *wrtc_data = SAFE_CALLOC(1, sizeof(webrtc_transport_data_t), webrtc_transport_data_t *);
  if (!wrtc_data) {
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate WebRTC transport data");
    return NULL;
  }
 
  // Create receive queue
  wrtc_data->recv_queue = ringbuffer_create(sizeof(webrtc_recv_msg_t), WEBRTC_RECV_QUEUE_SIZE);
  if (!wrtc_data->recv_queue) {
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_MEMORY, "Failed to create receive queue");
    return NULL;
  }
 
  // Initialize synchronization primitives
  if (mutex_init(&wrtc_data->queue_mutex, "webrtc_queue") != 0) {
    ringbuffer_destroy(wrtc_data->recv_queue);
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize queue mutex");
    return NULL;
  }
 
  if (cond_init(&wrtc_data->queue_cond, "queue") != 0) {
    mutex_destroy(&wrtc_data->queue_mutex);
    ringbuffer_destroy(wrtc_data->recv_queue);
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize queue condition variable");
    return NULL;
  }
 
  if (mutex_init(&wrtc_data->state_mutex, "webrtc_state") != 0) {
    cond_destroy(&wrtc_data->queue_cond);
    mutex_destroy(&wrtc_data->queue_mutex);
    ringbuffer_destroy(wrtc_data->recv_queue);
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize state mutex");
    return NULL;
  }
 
  if (mutex_init(&wrtc_data->send_mutex, "webrtc_send") != 0) {
    mutex_destroy(&wrtc_data->state_mutex);
    cond_destroy(&wrtc_data->queue_cond);
    mutex_destroy(&wrtc_data->queue_mutex);
    ringbuffer_destroy(wrtc_data->recv_queue);
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to initialize send mutex");
    return NULL;
  }
 
  // Initialize WebRTC data
  wrtc_data->peer_conn = peer_conn;
  wrtc_data->data_channel = data_channel;
  wrtc_data->is_connected = false; // Will be set to true in on_open callback
 
  // Register DataChannel callbacks
  webrtc_datachannel_callbacks_t callbacks = {
      .on_open = webrtc_on_open,
      .on_close = webrtc_on_close,
      .on_error = webrtc_on_error,
      .on_message = webrtc_on_message,
      .user_data = wrtc_data,
  };
 
  asciichat_error_t result = webrtc_datachannel_set_callbacks(data_channel, &callbacks);
  if (result != ASCIICHAT_OK) {
    mutex_destroy(&wrtc_data->send_mutex);
    mutex_destroy(&wrtc_data->state_mutex);
    cond_destroy(&wrtc_data->queue_cond);
    mutex_destroy(&wrtc_data->queue_mutex);
    ringbuffer_destroy(wrtc_data->recv_queue);
    SAFE_FREE(wrtc_data);
    SAFE_FREE(transport);
    SET_ERRNO(ERROR_INTERNAL, "Failed to set DataChannel callbacks");
    return NULL;
  }
 
  // IMPORTANT: The transport is always created from peer_manager's on_datachannel_open callback,
  // which means the DataChannel is ALREADY OPEN when we get here. However, by setting our own
  // callbacks above (webrtc_datachannel_set_callbacks), we replaced the callbacks that would
  // have set dc->is_open=true. So we need to manually mark both the transport AND the DataChannel
  // as open/connected now.
  //
  // We cannot rely on webrtc_on_open being called later because:
  // 1. The DataChannel is already open
  // 2. libdatachannel won't fire the open event again
  // 3. Setting callbacks after open doesn't trigger a retroactive open event
 
  // Mark DataChannel as open (needed for webrtc_datachannel_send() check)
  webrtc_datachannel_set_open_state(data_channel, true);
 
  // Mark transport as connected
  mutex_lock(&wrtc_data->state_mutex);
  wrtc_data->is_connected = true;
  mutex_unlock(&wrtc_data->state_mutex);
  log_debug("Transport and DataChannel marked as connected/open (already open from peer_manager callback)");
 
  // Initialize transport
  transport->methods = &webrtc_methods;
  transport->crypto_ctx = crypto_ctx;
  transport->impl_data = wrtc_data;
 
  log_info("Created WebRTC transport (crypto: %s)", crypto_ctx ? "enabled" : "disabled");
 
  // Register WebRTC transport implementation data with transport as parent
  NAMED_REGISTER(wrtc_data, "impl", "webrtc_impl", "0x%tx", (uintptr_t)(const void *)(transport));
 
  return transport;
}

References ASCIICHAT_OK, cond_destroy(), cond_init(), acip_transport::crypto_ctx, webrtc_transport_data_t::data_channel, ERROR_INTERNAL, ERROR_INVALID_PARAM, ERROR_MEMORY, acip_transport::impl_data, webrtc_transport_data_t::is_connected, log_debug, log_info, acip_transport::methods, mutex_destroy(), mutex_init(), mutex_lock, mutex_unlock, NAMED_REGISTER, webrtc_datachannel_callbacks_t::on_open, webrtc_transport_data_t::peer_conn, webrtc_transport_data_t::queue_cond, webrtc_transport_data_t::queue_mutex, webrtc_transport_data_t::recv_queue, ringbuffer_create(), ringbuffer_destroy(), SAFE_CALLOC, SAFE_FREE, SAFE_MALLOC, webrtc_transport_data_t::send_mutex, SET_ERRNO, webrtc_transport_data_t::state_mutex, webrtc_datachannel_set_callbacks(), webrtc_datachannel_set_open_state(), and WEBRTC_RECV_QUEUE_SIZE.

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.

Connects to a WebSocket server at the specified URL and creates a transport for sending/receiving ACIP packets over the WebSocket connection.

Parameters

name	Debug name for tracking this transport (required, e.g., "transport_websocket_client")
url	WebSocket URL (e.g., "ws://localhost:27225" or "wss://example.com/path")
crypto_ctx	Optional encryption context (can be NULL)

Returns

Transport instance or NULL on failure

Note

The URL must include the protocol (ws:// or wss://)

Port defaults to 80 for ws:// and 443 for wss:// if not specified

Connection is established synchronously during creation

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.