feat: support multiple juice_mux_listen callbacks

Supports multiple callbacks for unhandled STUN requests.

Stores multiple registries, one per local listening port in a
`conn_registry_t **registries;` field on `conn_mode_entry_t`.

The list of registries grows in the same way as the list of
agents in `conn_mode_entry_t`.

Adds a `port` property to `conn_registry_t` which is the port
the registry is bound to, and an `index` property which is the
index of the registry in the containing `registries` list in
`conn_mode_entry_t`.

Author: achingbrain

src/conn.c

@@ -33,30 +33,54 @@ typedef struct conn_mode_entry {
 	int (*get_addrs_func)(juice_agent_t *agent, addr_record_t *records, size_t size);
 
 	mutex_t mutex;
-	conn_registry_t *registry;
+	conn_registry_t **registries;
+	int registries_size;
+	int registries_count;
+
 } conn_mode_entry_t;
 
+typedef struct conn_impl {
+	conn_registry_t *registry;
+	timestamp_t next_timestamp;
+	bool finished;
+} conn_impl_t;
+
 #define MODE_ENTRIES_SIZE 3
 
 static conn_mode_entry_t mode_entries[MODE_ENTRIES_SIZE] = {
     {conn_poll_registry_init, conn_poll_registry_cleanup, conn_poll_init, conn_poll_cleanup,
      conn_poll_lock, conn_poll_unlock, conn_poll_interrupt, conn_poll_send, conn_poll_get_addrs,
-     MUTEX_INITIALIZER, NULL},
+     MUTEX_INITIALIZER, NULL, 0, 0},
     {conn_mux_registry_init, conn_mux_registry_cleanup, conn_mux_init, conn_mux_cleanup,
      conn_mux_lock, conn_mux_unlock, conn_mux_interrupt, conn_mux_send, conn_mux_get_addrs,
-     MUTEX_INITIALIZER, NULL},
+     MUTEX_INITIALIZER, NULL, 0, 0},
     {NULL, NULL, conn_thread_init, conn_thread_cleanup, conn_thread_lock, conn_thread_unlock,
-     conn_thread_interrupt, conn_thread_send, conn_thread_get_addrs, MUTEX_INITIALIZER, NULL}};
+     conn_thread_interrupt, conn_thread_send, conn_thread_get_addrs, MUTEX_INITIALIZER, NULL, 0, 0}};
 
 static conn_mode_entry_t *get_mode_entry(juice_agent_t *agent) {
 	juice_concurrency_mode_t mode = agent->config.concurrency_mode;
 	assert(mode >= 0 && mode < MODE_ENTRIES_SIZE);
 	return mode_entries + (int)mode;
 }
 
+static conn_registry_t *get_registry(conn_mode_entry_t *entry, uint16_t port) {
+	for (int i = 0; i < entry->registries_size; i++) {
+		if (!entry->registries[i]) {
+			continue;
+		}
+
+		if (entry->registries[i]->port == port) {
+			return entry->registries[i];
+		}
+	}
+
+	return NULL;
+}
+
 static int acquire_registry(conn_mode_entry_t *entry, udp_socket_config_t *config) {
 	// entry must be locked
-	conn_registry_t *registry = entry->registry;
+	conn_registry_t *registry = get_registry(entry, config->port_begin);
+
 	if (!registry) {
 		if (!entry->registry_init_func)
 			return 0;
@@ -91,8 +115,38 @@ static int acquire_registry(conn_mode_entry_t *entry, udp_socket_config_t *confi
 			return -1;
 		}
 
-		entry->registry = registry;
+		int i = 0;
+			while (i < entry->registries_size && entry->registries[i])
+				++i;
+
+		if (i == entry->registries_size) {
+			int new_size = entry->registries_size * 2;
+
+			if (new_size == 0) {
+				new_size = 1;
+			}
 
+			JLOG_DEBUG("Reallocating cb_mux_incomings array, new_size=%d", new_size);
+			assert(new_size > 0);
+
+			conn_registry_t **new_registries =
+					realloc(entry->registries, new_size * sizeof(conn_registry_t *));
+			if (!new_registries) {
+				JLOG_FATAL("Memory reallocation failed for cb_mux_incomings array");
+				mutex_unlock(&registry->mutex);
+				mutex_unlock(&entry->mutex);
+				return -1;
+			}
+
+			entry->registries = new_registries;
+			entry->registries_size = new_size;
+			memset(entry->registries + i, 0, (new_size - i) * sizeof(conn_registry_t *));
+		}
+
+		registry->index = i;
+		registry->port = config->port_begin;
+
+		entry->registries[i] = registry;
 	} else {
 		mutex_lock(&registry->mutex);
 	}
@@ -101,24 +155,31 @@ static int acquire_registry(conn_mode_entry_t *entry, udp_socket_config_t *confi
 	return 0;
 }
 
-static void release_registry(conn_mode_entry_t *entry) {
+static void release_registry(conn_mode_entry_t *entry, conn_registry_t *registry) {
 	// entry must be locked
-	conn_registry_t *registry = entry->registry;
 	if (!registry)
 		return;
 
 	// registry must be locked
-
 	if (registry->agents_count == 0 && registry->cb_mux_incoming == NULL) {
 		JLOG_DEBUG("No connection left, destroying connections registry");
 		mutex_unlock(&registry->mutex);
 
 		if (entry->registry_cleanup_func)
 			entry->registry_cleanup_func(registry);
 
+		if (registry->index > -1) {
+			int i = registry->index;
+			assert(entry->registries[i] == registry);
+			entry->registries[i] = NULL;
+			registry->index = -1;
+		}
+
+		assert(entry->registries_count > 0);
+		--entry->registries_count;
+
 		free(registry->agents);
 		free(registry);
-		entry->registry = NULL;
 		return;
 	}
 
@@ -136,7 +197,7 @@ int conn_create(juice_agent_t *agent, udp_socket_config_t *config) {
 		return -1;
 	}
 
-	conn_registry_t *registry = entry->registry;
+	conn_registry_t *registry = get_registry(entry, config->port_begin);
 
 	JLOG_DEBUG("Creating connection");
 	if (registry) {
@@ -164,7 +225,7 @@ int conn_create(juice_agent_t *agent, udp_socket_config_t *config) {
 		}
 
 		if (get_mode_entry(agent)->init_func(agent, registry, config)) {
-			release_registry(entry); // unlocks the registry
+			release_registry(entry, registry); // unlocks the registry
 			mutex_unlock(&entry->mutex);
 			return -1;
 		}
@@ -194,7 +255,9 @@ void conn_destroy(juice_agent_t *agent) {
 	mutex_lock(&entry->mutex);
 
 	JLOG_DEBUG("Destroying connection");
-	conn_registry_t *registry = entry->registry;
+	conn_impl_t *conn_impl = agent->conn_impl;
+
+	conn_registry_t *registry = conn_impl->registry;
 	if (registry) {
 		mutex_lock(&registry->mutex);
 
@@ -210,7 +273,7 @@ void conn_destroy(juice_agent_t *agent) {
 		assert(registry->agents_count > 0);
 		--registry->agents_count;
 
-		release_registry(entry); // unlocks the registry
+		release_registry(entry, registry); // unlocks the registry
 
 	} else {
 		entry->cleanup_func(agent);
@@ -264,7 +327,7 @@ static int juice_mux_stop_listen(const char *bind_address, int local_port) {
 
 	mutex_lock(&entry->mutex);
 
-	conn_registry_t *registry = entry->registry;
+	conn_registry_t *registry = get_registry(entry, local_port);
 	if (!registry) {
 		mutex_unlock(&entry->mutex);
 		return -1;
@@ -276,7 +339,7 @@ static int juice_mux_stop_listen(const char *bind_address, int local_port) {
 	registry->mux_incoming_user_ptr = NULL;
 	conn_mux_interrupt_registry(registry);
 
-	release_registry(entry);
+	release_registry(entry, registry);
 
 	mutex_unlock(&entry->mutex);
 
@@ -296,7 +359,7 @@ int juice_mux_listen(const char *bind_address, int local_port, juice_cb_mux_inco
 
 	mutex_lock(&entry->mutex);
 
-	if (entry->registry) {
+	if (get_registry(entry, local_port)) {
 		mutex_unlock(&entry->mutex);
 		JLOG_DEBUG("juice_mux_listen needs to be called before establishing any mux connection.");
 		return -1;
@@ -307,7 +370,7 @@ int juice_mux_listen(const char *bind_address, int local_port, juice_cb_mux_inco
 		return -1;
 	}
 
-	conn_registry_t *registry = entry->registry;
+	conn_registry_t *registry = get_registry(entry, local_port);
 	if(!registry) {
 		mutex_unlock(&entry->mutex);
 		return -1;

src/conn.h

@@ -25,6 +25,8 @@ typedef struct juice_agent juice_agent_t;
 // See include/juice/juice.h for implemented concurrency modes
 
 typedef struct conn_registry {
+	int index;
+	uint16_t port;
 	void *impl;
 	mutex_t mutex;
 	juice_agent_t **agents;