Simplify adding new swclt_wss_stop() API.

Author: andywolk

inc/signalwire-client-c/transport/websocket.h

@@ -97,6 +97,7 @@ SWCLT_DECLARE(ks_status_t) swclt_wss_connect(
 	uint32_t timeout_ms,
 	const SSL_CTX *ssl);
 
+SWCLT_DECLARE(void) swclt_wss_stop(swclt_wss_t *wss);
 SWCLT_DECLARE(void) swclt_wss_destroy(swclt_wss_t **wss);
 
 SWCLT_DECLARE(ks_status_t) swclt_wss_write(swclt_wss_t *wss, char *data);

src/connection.c

@@ -672,22 +672,16 @@ SWCLT_DECLARE(void) swclt_conn_destroy(swclt_conn_t **conn)
 		if ((*conn)->blade_connect_rpl) {
 			BLADE_CONNECT_RPL_DESTROY(&(*conn)->blade_connect_rpl);
 		}
-		/* Stop the websocket reader thread FIRST to prevent new jobs from being
-		 * added to the incoming_frame_pool. The reader thread calls on_incoming_frame
-		 * which adds jobs to the pool, so we must stop it before destroying the pool. */
-		if ((*conn)->wss && (*conn)->wss->reader_thread) {
-			ks_thread_request_stop((*conn)->wss->reader_thread);
-			ks_thread_join((*conn)->wss->reader_thread);
-			ks_thread_destroy(&(*conn)->wss->reader_thread);
-		}
-		/* Now destroy the incoming frame thread pool. This waits for all pending
-		 * jobs to complete. These jobs may call swclt_wss_write() which uses the
-		 * websocket mutex, so we must do this BEFORE destroying the websocket. */
+
+		/* 1. Stop websocket reader: prevents new jobs from being queued
+		 * 2. Drain thread pool: lets pending jobs finish (they use wss_mutex)
+		 * 3. Destroy websocket: now safe to free the mutex
+		 */
+
+		swclt_wss_stop((*conn)->wss);
 		if ((*conn)->incoming_frame_pool) {
 			ks_thread_pool_destroy(&(*conn)->incoming_frame_pool);
 		}
-		/* Now safe to destroy the websocket - no threads are using it anymore.
-		 * swclt_wss_destroy will see reader_thread already stopped/joined. */
 		swclt_wss_destroy(&(*conn)->wss);
 		ttl_tracker_destroy(&(*conn)->ttl);
 		ks_hash_destroy(&(*conn)->outstanding_requests);

src/transport/websocket.c

@@ -312,19 +312,22 @@ static ks_status_t __connect_socket(swclt_wss_t *ctx)
 	return status;
 }
 
+SWCLT_DECLARE(void) swclt_wss_stop(swclt_wss_t *wss)
+{
+	if (wss && wss->reader_thread) {
+		ks_log(KS_LOG_DEBUG, "Stopping websocket reader thread");
+		ks_thread_request_stop(wss->reader_thread);
+		ks_thread_join(wss->reader_thread);
+		ks_thread_destroy(&wss->reader_thread);
+	}
+}
+
 SWCLT_DECLARE(void) swclt_wss_destroy(swclt_wss_t **wss)
 {
 	if (wss && *wss) {
 		ks_pool_t *pool = (*wss)->pool;
 		ks_log(KS_LOG_INFO, "Shutting down websocket");
-		if ((*wss)->reader_thread) {
-			ks_thread_request_stop((*wss)->reader_thread);
-		}
-
-		if ((*wss)->reader_thread) {
-			ks_thread_join((*wss)->reader_thread);
-			ks_thread_destroy(&(*wss)->reader_thread);
-		}
+		swclt_wss_stop(*wss);
 		if ((*wss)->wss_mutex) {
 			ks_mutex_destroy(&(*wss)->wss_mutex);
 		}