webrtc/transport.c

Go to the documentation of this file.

 
#include "network/acip/transport.h"
#include "network/webrtc/webrtc.h"
#include "log/logging.h"
#include "ringbuffer.h"
#include "platform/mutex.h"
#include "platform/cond.h"
#include <string.h>
 
#define WEBRTC_RECV_QUEUE_SIZE 64
 
typedef struct {
  uint8_t *data; 
  size_t len;    
} webrtc_recv_msg_t;
 
typedef struct {
  webrtc_peer_connection_t *peer_conn; 
  webrtc_data_channel_t *data_channel; 
  ringbuffer_t *recv_queue;            
  mutex_t queue_mutex;                 
  cond_t queue_cond;                   
  bool is_connected;                   
  mutex_t state_mutex;                 
} webrtc_transport_data_t;
 
// =============================================================================
// DataChannel Callbacks
// =============================================================================
 
static void webrtc_on_message(webrtc_data_channel_t *channel, const uint8_t *data, size_t len, void *user_data) {
  (void)channel; // Unused
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)user_data;
 
  if (!wrtc || !data || len == 0) {
    log_error("Invalid DataChannel message callback parameters");
    return;
  }
 
  // Allocate message structure
  webrtc_recv_msg_t msg;
  msg.data = SAFE_MALLOC(len, uint8_t *);
  if (!msg.data) {
    log_error("Failed to allocate receive message buffer (%zu bytes)", len);
    return;
  }
 
  memcpy(msg.data, data, len);
  msg.len = len;
 
  // Push to receive queue (thread-safe)
  mutex_lock(&wrtc->queue_mutex);
 
  bool success = ringbuffer_write(wrtc->recv_queue, &msg);
  if (!success) {
    // Queue full - drop oldest message to make room
    webrtc_recv_msg_t dropped_msg;
    if (ringbuffer_read(wrtc->recv_queue, &dropped_msg)) {
      log_warn("Receive queue full, dropping oldest message (%zu bytes)", dropped_msg.len);
      SAFE_FREE(dropped_msg.data);
    }
 
    // Try again
    success = ringbuffer_write(wrtc->recv_queue, &msg);
    if (!success) {
      log_error("Failed to write to receive queue after drop");
      SAFE_FREE(msg.data);
      mutex_unlock(&wrtc->queue_mutex);
      return;
    }
  }
 
  // Signal waiting recv() call
  cond_signal(&wrtc->queue_cond);
  mutex_unlock(&wrtc->queue_mutex);
 
  log_debug_every(1000000, "WebRTC message received (%zu bytes), queue size: %zu", len,
                  ringbuffer_size(wrtc->recv_queue));
}
 
static void webrtc_on_open(webrtc_data_channel_t *channel, void *user_data) {
  (void)channel;
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)user_data;
 
  if (!wrtc) {
    return;
  }
 
  mutex_lock(&wrtc->state_mutex);
  wrtc->is_connected = true;
  mutex_unlock(&wrtc->state_mutex);
 
  log_info("WebRTC DataChannel opened, transport ready");
}
 
static void webrtc_on_error(webrtc_data_channel_t *channel, const char *error_msg, void *user_data) {
  (void)channel;
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)user_data;
 
  log_error("WebRTC DataChannel error: %s", error_msg ? error_msg : "unknown error");
 
  if (!wrtc) {
    return;
  }
 
  mutex_lock(&wrtc->state_mutex);
  wrtc->is_connected = false;
  mutex_unlock(&wrtc->state_mutex);
 
  // Wake any blocking recv() calls
  cond_broadcast(&wrtc->queue_cond);
}
 
static void webrtc_on_close(webrtc_data_channel_t *channel, void *user_data) {
  (void)channel;
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)user_data;
 
  log_info("WebRTC DataChannel closed");
 
  if (!wrtc) {
    return;
  }
 
  mutex_lock(&wrtc->state_mutex);
  wrtc->is_connected = false;
  mutex_unlock(&wrtc->state_mutex);
 
  // Wake any blocking recv() calls
  cond_broadcast(&wrtc->queue_cond);
}
 
// =============================================================================
// WebRTC Transport Methods
// =============================================================================
 
static asciichat_error_t webrtc_send(acip_transport_t *transport, const void *data, size_t len) {
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)transport->impl_data;
 
  mutex_lock(&wrtc->state_mutex);
  bool connected = wrtc->is_connected;
  mutex_unlock(&wrtc->state_mutex);
 
  if (!connected) {
    return SET_ERRNO(ERROR_NETWORK, "WebRTC transport not connected");
  }
 
  // Send via DataChannel
  asciichat_error_t result = webrtc_datachannel_send(wrtc->data_channel, data, len);
  if (result != ASCIICHAT_OK) {
    return SET_ERRNO(ERROR_NETWORK, "Failed to send on WebRTC DataChannel");
  }
 
  log_debug_every(1000000, "WebRTC sent %zu bytes", len);
  return ASCIICHAT_OK;
}
 
static asciichat_error_t webrtc_recv(acip_transport_t *transport, void **buffer, size_t *out_len,
                                     void **out_allocated_buffer) {
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)transport->impl_data;
 
  mutex_lock(&wrtc->queue_mutex);
 
  // Block until message arrives or connection closes
  while (ringbuffer_is_empty(wrtc->recv_queue)) {
    mutex_lock(&wrtc->state_mutex);
    bool connected = wrtc->is_connected;
    mutex_unlock(&wrtc->state_mutex);
 
    if (!connected) {
      mutex_unlock(&wrtc->queue_mutex);
      return SET_ERRNO(ERROR_NETWORK, "Connection closed while waiting for data");
    }
 
    // Wait for message arrival or connection close
    cond_wait(&wrtc->queue_cond, &wrtc->queue_mutex);
  }
 
  // Read message from queue
  webrtc_recv_msg_t msg;
  bool success = ringbuffer_read(wrtc->recv_queue, &msg);
  mutex_unlock(&wrtc->queue_mutex);
 
  if (!success) {
    return SET_ERRNO(ERROR_NETWORK, "Failed to read from receive queue");
  }
 
  // Return message to caller (caller owns the buffer)
  *buffer = msg.data;
  *out_len = msg.len;
  *out_allocated_buffer = msg.data;
 
  log_debug_every(1000000, "WebRTC received %zu bytes, queue remaining: %zu", msg.len,
                  ringbuffer_size(wrtc->recv_queue));
 
  return ASCIICHAT_OK;
}
 
static asciichat_error_t webrtc_close(acip_transport_t *transport) {
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)transport->impl_data;
 
  mutex_lock(&wrtc->state_mutex);
 
  if (!wrtc->is_connected) {
    mutex_unlock(&wrtc->state_mutex);
    return ASCIICHAT_OK; // Already closed
  }
 
  wrtc->is_connected = false;
  mutex_unlock(&wrtc->state_mutex);
 
  // Close DataChannel
  if (wrtc->data_channel) {
    webrtc_datachannel_close(wrtc->data_channel);
  }
 
  // Close peer connection
  if (wrtc->peer_conn) {
    webrtc_peer_connection_close(wrtc->peer_conn);
  }
 
  // Wake any blocking recv() calls
  cond_broadcast(&wrtc->queue_cond);
 
  log_debug("WebRTC transport closed");
  return ASCIICHAT_OK;
}
 
static acip_transport_type_t webrtc_get_type(acip_transport_t *transport) {
  (void)transport;
  return ACIP_TRANSPORT_WEBRTC;
}
 
static socket_t webrtc_get_socket(acip_transport_t *transport) {
  (void)transport;
  return INVALID_SOCKET_VALUE; // WebRTC has no underlying socket
}
 
static bool webrtc_is_connected(acip_transport_t *transport) {
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)transport->impl_data;
 
  mutex_lock(&wrtc->state_mutex);
  bool connected = wrtc->is_connected;
  mutex_unlock(&wrtc->state_mutex);
 
  return connected;
}
 
// =============================================================================
// WebRTC Transport Destroy Implementation
// =============================================================================
 
static void webrtc_destroy_impl(acip_transport_t *transport) {
  if (!transport || !transport->impl_data) {
    return;
  }
 
  webrtc_transport_data_t *wrtc = (webrtc_transport_data_t *)transport->impl_data;
 
  // Destroy peer connection and data channel
  if (wrtc->data_channel) {
    webrtc_datachannel_destroy(wrtc->data_channel);
    wrtc->data_channel = NULL;
  }
 
  if (wrtc->peer_conn) {
    webrtc_peer_connection_destroy(wrtc->peer_conn);
    wrtc->peer_conn = NULL;
  }
 
  // Clear receive queue and free buffered messages
  if (wrtc->recv_queue) {
    mutex_lock(&wrtc->queue_mutex);
 
    webrtc_recv_msg_t msg;
    while (ringbuffer_read(wrtc->recv_queue, &msg)) {
      if (msg.data) {
        SAFE_FREE(msg.data);
      }
    }
 
    mutex_unlock(&wrtc->queue_mutex);
    ringbuffer_destroy(wrtc->recv_queue);
    wrtc->recv_queue = NULL;
  }
 
  // Destroy synchronization primitives
  mutex_destroy(&wrtc->state_mutex);
  cond_destroy(&wrtc->queue_cond);
  mutex_destroy(&wrtc->queue_mutex);
 
  log_debug("Destroyed WebRTC transport resources");
}
 
// =============================================================================
// WebRTC Transport Method Table
// =============================================================================
 
static const acip_transport_methods_t webrtc_methods = {
    .send = webrtc_send,
    .recv = webrtc_recv,
    .close = webrtc_close,
    .get_type = webrtc_get_type,
    .get_socket = webrtc_get_socket,
    .is_connected = webrtc_is_connected,
    .destroy_impl = webrtc_destroy_impl,
};
 
// =============================================================================
// WebRTC Transport Creation
// =============================================================================
 
acip_transport_t *acip_webrtc_transport_create(webrtc_peer_connection_t *peer_conn, webrtc_data_channel_t *data_channel,
                                               crypto_context_t *crypto_ctx) {
  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_MALLOC(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) != 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) != 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) != 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;
  }
 
  // 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->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;
  }
 
  // 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");
 
  return transport;
}