From e97fd6b0de76b0038ee49e14b78d5da806aae434 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:34:27 -0600
Subject: [PATCH 1/8] fix: harden relay connection handling

---
 config.c             |  22 ++-
 main.c               |  29 +++-
 rtlmux.c             | 347 +++++++++++++++++++++++++------------------
 rtlmux.h             |   9 +-
 slog.c               |   6 +-
 tests/integration.py | 205 +++++++++++++++++++++++++
 6 files changed, 459 insertions(+), 159 deletions(-)
 create mode 100644 tests/integration.py

diff --git a/config.c b/config.c
index 34b8d56..98acae7 100644
--- a/config.c
+++ b/config.c
@@ -16,9 +16,22 @@ struct config config;
 static struct gengetopt_args_info args;
 
 int convertConfig(struct gengetopt_args_info *args) {
+  if(args->port_arg < 1 || args->port_arg > UINT16_MAX) {
+    fprintf(stderr, "rtl_tcp port must be between 1 and %u.\n", UINT16_MAX);
+    return 0;
+  }
+  if(args->listen_arg < 1 || args->listen_arg >= UINT16_MAX) {
+    fprintf(stderr, "Listening port must be between 1 and %u to reserve the following port for HTTP status.\n", UINT16_MAX - 1);
+    return 0;
+  }
+  if(args->host_arg == NULL || args->host_arg[0] == '\0') {
+    fprintf(stderr, "rtl_tcp host address cannot be empty.\n");
+    return 0;
+  }
+
   config.host = args->host_arg;
-  config.port = args->port_arg;
-  config.clientPort = args->listen_arg;
+  config.port = (uint16_t)args->port_arg;
+  config.clientPort = (uint16_t)args->listen_arg;
   config.delayed = args->delayed_flag;
   config.restart = args->restart_flag;
 
@@ -42,5 +55,8 @@ int parseConfig(int argc, char **argv) {
     exit(4);
   }
   
-  return convertConfig(&args);
+  if(!convertConfig(&args))
+    exit(4);
+
+  return 1;
 }
diff --git a/main.c b/main.c
index 45ef52e..6788ab8 100644
--- a/main.c
+++ b/main.c
@@ -3,34 +3,53 @@
 
 #include "rtlmux.h"
 
+#include <errno.h>
+#include <pthread.h>
 #include <signal.h>
+#include <string.h>
 
-volatile unsigned char timeToExit = 0;
+volatile sig_atomic_t timeToExit = 0;
 
 void signalExit(int sig) {
+  (void)sig;
   timeToExit = 1;
 }
 
 int main(int argc, char **argv) {
   pthread_t threadServer;
+  int result;
   
   parseConfig(argc, argv);
   slog_init(NULL, NULL, LOG_EXTRA, LOG_DEBUG, 1);
 
   struct sigaction sigact;
+  memset(&sigact, 0, sizeof(sigact));
   sigact.sa_handler = signalExit;
   sigact.sa_flags = 0;
-  sigaction(SIGTERM, &sigact, NULL);
-  sigaction(SIGINT, &sigact, NULL);
+  sigemptyset(&sigact.sa_mask);
+  if(sigaction(SIGTERM, &sigact, NULL) != 0 || sigaction(SIGINT, &sigact, NULL) != 0) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not configure signal handlers: %s", strerror(errno));
+    return 1;
+  }
   
   do {
-    pthread_create(&threadServer, NULL, serverThread, NULL);
+    result = pthread_create(&threadServer, NULL, serverThread, NULL);
+    if(result != 0) {
+      slog(LOG_FATAL, SLOG_FATAL, "Could not start server thread: %s", strerror(result));
+      return 1;
+    }
 
-    pthread_join(threadServer, NULL);
+    result = pthread_join(threadServer, NULL);
+    if(result != 0) {
+      slog(LOG_FATAL, SLOG_FATAL, "Could not join server thread: %s", strerror(result));
+      return 1;
+    }
 
     if (timeToExit == 2) {
       slog(LOG_INFO, SLOG_INFO, "Restarting.");
       timeToExit = 0;
     }
   } while (timeToExit != 1);
+
+  return 0;
 }
diff --git a/rtlmux.c b/rtlmux.c
index 525250f..820c7ef 100644
--- a/rtlmux.c
+++ b/rtlmux.c
@@ -1,6 +1,7 @@
 #define _GNU_SOURCE
 #include <stdlib.h>
 #include <stdint.h>
+#include <inttypes.h>
 #include <unistd.h>
 #include <string.h>
 #include <errno.h>
@@ -28,30 +29,17 @@
 
 #include <sys/queue.h>
 
-// Define this to enable thread safety around the lists
-//#define THREADED
-
-#ifndef THREADED
-#define pthread_rwlock_wrlock(a)
-#define pthread_rwlock_rdlock(a)
-#define pthread_rwlock_unlock(a)
-#endif
-
 struct event_base *event_base = NULL;
 struct bufferevent *serverConnection = NULL;
 
 unsigned long dataBlocks = 0;
 unsigned long dataBlocksSize = 0;
 struct rtlData {
-  LIST_ENTRY(rtlData) next;
   uint32_t references;
   uint32_t len;
   uint8_t *data;
 };
-static LIST_HEAD(rtlDataHead, rtlData) rtlDataList = LIST_HEAD_INITIALIZER(rtlDataList);
-static pthread_rwlock_t rtlDataLock;
 
-#define CLIENT_UNKNOWN 1
 #define CLIENT_READY 2
 #define CLIENT_INIT 4
 struct client {
@@ -73,21 +61,20 @@ struct client {
 };
 
 static LIST_HEAD(clienthead, client) clients = LIST_HEAD_INITIALIZER(clients);
-static pthread_rwlock_t clientLock;
 
-static struct client *addClient(struct bufferevent *bev, void *ptr) {
+static struct client *addClient(struct bufferevent *bev) {
   uint32_t clientFlags = CLIENT_INIT;
   
   struct client *client = (struct client *)calloc(1, sizeof(struct client));
+  if(client == NULL)
+    return NULL;
   
   client->bev = bev;
   client->flags = clientFlags;
   client->data.in = client->data.out = 0;
   client->connected = time(NULL);
   
-  pthread_rwlock_wrlock(&clientLock);
   LIST_INSERT_HEAD(&clients, client, peer);
-  pthread_rwlock_unlock(&clientLock);
   
   return client;
 }
@@ -96,19 +83,21 @@ static void removeClient(struct client *client) {
   if(!client)
     return;
   
-  pthread_rwlock_wrlock(&clientLock);
   LIST_REMOVE(client, peer);
   free(client);
-  pthread_rwlock_unlock(&clientLock);
+
+  if(config.delayed && LIST_EMPTY(&clients)) {
+    slog(LOG_INFO, SLOG_INFO, "Last user disconnected.");
+    timeToExit = config.restart ? 2 : 1;
+  }
 }
 
 void releaseDataRef(const void *d, unsigned long len, void *ptr) {
+  (void)d;
+  (void)len;
   struct rtlData *data = (struct rtlData *)ptr;
   --data->references;
   if(data->references == 0) {
-    //pthread_rwlock_wrlock(&rtlDataLock);
-    //LIST_REMOVE(data, next);
-    //pthread_rwlock_unlock(&rtlDataLock);
     dataBlocks--;
     dataBlocksSize -= data->len;
     free(data); // This is a single malloc for both the data and header
@@ -117,7 +106,6 @@ void releaseDataRef(const void *d, unsigned long len, void *ptr) {
 
 int sendDataToAllClients(struct rtlData *data) {
   struct client *client;
-  pthread_rwlock_rdlock(&clientLock);
   LIST_FOREACH(client, &clients, peer) {
     if(client->flags == CLIENT_READY) {
       struct evbuffer *ev = bufferevent_get_output(client->bev);
@@ -126,25 +114,18 @@ int sendDataToAllClients(struct rtlData *data) {
         client->data.droppedCount ++;
         continue;
       }
-      ++data->references;
-      evbuffer_add_reference(ev, data->data, data->len, releaseDataRef, data);
-      client->data.out += data->len;
+      if(evbuffer_add_reference(ev, data->data, data->len, releaseDataRef, data) == 0) {
+        ++data->references;
+        client->data.out += data->len;
+      } else {
+        client->data.dropped += data->len;
+        client->data.droppedCount ++;
+      }
     }
   }
-  pthread_rwlock_unlock(&clientLock);
   return data->references;
 }
 
-void sendToAllClients(char *buf, size_t len, uint32_t flags) {
-  struct client *client;
-  pthread_rwlock_rdlock(&clientLock);
-  LIST_FOREACH(client, &clients, peer) {
-    if((client->flags & flags) != 0)
-      bufferevent_write(client->bev, buf, len);
-  }
-  pthread_rwlock_unlock(&clientLock);
-}
-
 static void logCB(int severity, const char *msg) {
   int level;
   int flag;
@@ -156,7 +137,7 @@ static void logCB(int severity, const char *msg) {
     default: level = LOG_LIVE; flag = LOG_LIVE; break;
   }
   
-  slog(level, flag, msg);
+  slog(level, flag, "%s", msg);
 }
 
 struct serverInfo {
@@ -165,8 +146,8 @@ struct serverInfo {
   uint32_t tuner_type;
   uint32_t tuner_gain_count;
   struct {
-    unsigned int value;
-    unsigned char set;
+    uint32_t value;
+    uint8_t set;
   } params[0xd]; // Store all the parameters as a simple command array
   struct {
     uint64_t in;
@@ -174,52 +155,106 @@ struct serverInfo {
   } data;
 } serverInfo;
 
+static void readyClient(struct client *client) {
+  uint8_t header[12];
+  memcpy(header, serverInfo.magic, 4);
+  memcpy(header + 4, &serverInfo.tuner_type, 4);
+  memcpy(header + 8, &serverInfo.tuner_gain_count, 4);
+
+  if(bufferevent_write(client->bev, header, sizeof(header)) == 0) {
+    client->data.out += 12;
+    client->flags = CLIENT_READY;
+  }
+}
+
+static void readyWaitingClients(void) {
+  struct client *client;
+  LIST_FOREACH(client, &clients, peer) {
+    if(client->flags == CLIENT_INIT)
+      readyClient(client);
+  }
+}
+
 static void serverErrorEventCB(struct bufferevent *, short, void *);
 static void serverReadCB(struct bufferevent *, void *);
+static void connectToServerSoon(void);
+
+static void disconnectServer(struct bufferevent *bev) {
+  if(serverConnection == bev)
+    serverConnection = NULL;
+  bufferevent_free(bev);
+}
+
+static void connectToServer(void) {
+  if(serverConnection != NULL || timeToExit)
+    return;
 
-static void connectToServer(void *arg) {
-  struct bufferevent **serverConnection = (struct bufferevent **)arg;
   slog(LOG_INFO, SLOG_INFO, "Starting connection lookup for %s:%d", config.host, config.port);
-  *serverConnection = bufferevent_socket_new(event_base, -1, BEV_OPT_CLOSE_ON_FREE);
-  bufferevent_socket_connect_hostname(*serverConnection, NULL, AF_UNSPEC, config.host, config.port);
+  serverConnection = bufferevent_socket_new(event_base, -1, BEV_OPT_CLOSE_ON_FREE);
+  if(serverConnection == NULL) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not allocate server connection.");
+    timeToExit = 1;
+    return;
+  }
+
+  bufferevent_setcb(serverConnection, serverReadCB, NULL, serverErrorEventCB, NULL);
+  bufferevent_setwatermark(serverConnection, EV_READ, 1, 0);
+  bufferevent_enable(serverConnection, EV_READ|EV_WRITE);
+  if(bufferevent_socket_connect_hostname(serverConnection, NULL, AF_UNSPEC, config.host, config.port) != 0) {
+    slog(LOG_ERROR, SLOG_ERROR, "Could not start connection to %s:%d", config.host, config.port);
+    disconnectServer(serverConnection);
+    connectToServerSoon();
+    return;
+  }
   slog(LOG_INFO, SLOG_INFO, "Started to connect to %s:%d", config.host, config.port);
-  bufferevent_setcb(*serverConnection, serverReadCB, NULL, serverErrorEventCB, serverConnection);
-  bufferevent_setwatermark(*serverConnection, EV_READ, 16384, 0);
-  bufferevent_enable(*serverConnection, EV_READ|EV_WRITE);
 }
 
 static void connectToServerCB(int a, short b, void *arg) {
-  connectToServer(arg);
+  (void)a;
+  (void)b;
+  (void)arg;
+  connectToServer();
 }
 
-static void connectToServerSoon(void *ctx) {
-  struct event *ev;
+static void connectToServerSoon(void) {
   struct timeval tv;
 
+  if(timeToExit)
+    return;
+
   tv.tv_sec = 1;
   tv.tv_usec = 0;
 
-  ev = evtimer_new(event_base, connectToServerCB, ctx);
-  evtimer_add(ev, &tv);
+  if(event_base_once(event_base, -1, EV_TIMEOUT, connectToServerCB, NULL, &tv) != 0) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not schedule server reconnect.");
+    timeToExit = 1;
+  }
 }
 
 static void serverReadCB(struct bufferevent *bev, void *ctx) {
+  (void)ctx;
   struct rtlData *data;
+  struct evbuffer *ev = bufferevent_get_input(bev);
   
   if(serverInfo.state == SERVER_NEW) {
-    serverInfo.data.in += bufferevent_read(bev, serverInfo.magic, 4);
-    serverInfo.data.in += bufferevent_read(bev, &serverInfo.tuner_type, 4);
-    serverInfo.data.in += bufferevent_read(bev, &serverInfo.tuner_gain_count, 4);
+    if(evbuffer_get_length(ev) < 12)
+      return;
+
+    serverInfo.data.in += evbuffer_remove(ev, serverInfo.magic, 4);
+    serverInfo.data.in += evbuffer_remove(ev, &serverInfo.tuner_type, 4);
+    serverInfo.data.in += evbuffer_remove(ev, &serverInfo.tuner_gain_count, 4);
     if(serverInfo.magic[0] == 'R' && serverInfo.magic[1] == 'T' && serverInfo.magic[2] == 'L' && serverInfo.magic[3] == '0') {
       serverInfo.state = SERVER_CONNECTED;
       slog(LOG_INFO, SLOG_INFO, "Connected to server.");
+      readyWaitingClients();
     } else { // Failed to receive the magic header
       slog(LOG_ERROR, SLOG_ERROR, "Failed to receive magic header from server.");
-      bufferevent_free(bev);
+      disconnectServer(bev);
+      serverInfo.state = SERVER_NEW;
       if (config.delayed) {
         timeToExit = config.restart ? 2 : 1;
       } else {
-        connectToServerSoon(ctx);
+        connectToServerSoon();
       }
       return;
     }
@@ -228,16 +263,15 @@ static void serverReadCB(struct bufferevent *bev, void *ctx) {
     for(i = 0; i < 0xd; i++) {
       if(serverInfo.params[i].set) {
         struct command cmd;
-        cmd.cmd = i+1;
+        cmd.cmd = (uint8_t)(i+1);
         cmd.param = serverInfo.params[i].value;
-        slog(LOG_INFO, SLOG_INFO, "Sending command %d with param %lu", cmd.cmd, ntohl(cmd.param));
+        slog(LOG_INFO, SLOG_INFO, "Sending command %d with param %u", cmd.cmd, ntohl(cmd.param));
         serverInfo.data.out += sizeof(cmd);
         bufferevent_write(bev, &cmd, sizeof(cmd));
       }
     }
   }
   
-  struct evbuffer *ev = bufferevent_get_input(bev);
   size_t availLen = evbuffer_get_length(ev);
   
   if(availLen == 0) // We may not have data, so return
@@ -247,42 +281,45 @@ static void serverReadCB(struct bufferevent *bev, void *ctx) {
     availLen = 256*1024; // Limit our input sizes to 256k chunks
 
   data = (struct rtlData *)malloc(sizeof(struct rtlData) + availLen);
+  if(data == NULL) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not allocate input data buffer.");
+    timeToExit = 1;
+    return;
+  }
   memset(data, 0, sizeof(struct rtlData));
-  data->data = (void *)data + sizeof(struct rtlData);
+  data->data = (uint8_t *)data + sizeof(struct rtlData);
   data->references = 0;
-  serverInfo.data.in += data->len = bufferevent_read(bev, data->data, availLen);
+  data->len = (uint32_t)bufferevent_read(bev, data->data, availLen);
+  serverInfo.data.in += data->len;
   
   if(sendDataToAllClients(data) == 0) {
     // No one was listening
     free(data);
-    if (config.delayed) {
-      slog(LOG_INFO, SLOG_INFO, "Last user disconnected.");
-      timeToExit = config.restart ? 2 : 1;
-    }
   } else {
     dataBlocks++;
     dataBlocksSize += data->len;    
-    // Track the data block
-    //pthread_rwlock_wrlock(&rtlDataLock);
-    //LIST_INSERT_HEAD(&rtlDataList, data, next);    
-    //pthread_rwlock_unlock(&rtlDataLock);
   }
 }
 
 static void serverErrorEventCB(struct bufferevent *bev, short events, void *ctx) {
-  if (events & BEV_EVENT_ERROR)
-    slog(LOG_ERROR, SLOG_ERROR, "Error from server side bufferevent: %s", strerror(errno));
+  (void)ctx;
+  if (events & BEV_EVENT_ERROR) {
+    int error = EVUTIL_SOCKET_ERROR();
+    slog(LOG_ERROR, SLOG_ERROR, "Error from server side bufferevent: %s", evutil_socket_error_to_string(error));
+  }
   if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
-    bufferevent_free(bev);
+    disconnectServer(bev);
     slog(LOG_INFO, SLOG_INFO, "Disconnecting server.");
     serverInfo.state = SERVER_NEW;
-    connectToServerSoon(ctx);
+    connectToServerSoon();
   }
 }
 
 static void errorEventCB(struct bufferevent *bev, short events, void *ctx) {
-  if (events & BEV_EVENT_ERROR)
-    slog(LOG_ERROR, SLOG_ERROR, "Error from bufferevent: %s", strerror(errno));
+  if (events & BEV_EVENT_ERROR) {
+    int error = EVUTIL_SOCKET_ERROR();
+    slog(LOG_ERROR, SLOG_ERROR, "Error from bufferevent: %s", evutil_socket_error_to_string(error));
+  }
   if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
     struct client *client = (struct client *)ctx;
     char ipBuf[128];
@@ -299,11 +336,18 @@ static void errorEventCB(struct bufferevent *bev, short events, void *ctx) {
 }
 
 void serverSendCommand(struct command cmd) {
+  if(cmd.cmd < 1 || cmd.cmd > 0xd) {
+    slog(LOG_WARN, SLOG_WARN, "Ignoring invalid command: %u", cmd.cmd);
+    return;
+  }
+
   serverInfo.params[cmd.cmd-1].value = cmd.param;
   serverInfo.params[cmd.cmd-1].set = 1;
-  slog(LOG_LIVE, SLOG_DEBUG, "Sending command to server: %d: %lu", cmd.cmd, ntohl(cmd.param));
-  serverInfo.data.out += sizeof(cmd);
-  bufferevent_write(serverConnection, &cmd, sizeof(cmd));
+  if(serverConnection != NULL && serverInfo.state == SERVER_CONNECTED) {
+    slog(LOG_LIVE, SLOG_DEBUG, "Sending command to server: %d: %u", cmd.cmd, ntohl(cmd.param));
+    serverInfo.data.out += sizeof(cmd);
+    bufferevent_write(serverConnection, &cmd, sizeof(cmd));
+  }
 }
 
 #define RTL_FREQUENCY 0x01
@@ -322,61 +366,62 @@ void serverSendCommand(struct command cmd) {
 
 static void clientReadCB(struct bufferevent *bev, void *ctx) {
   struct command cmd;
-  size_t l;
   struct client *client = (struct client *)ctx;
-  while((l = bufferevent_read(bev, &cmd, sizeof(cmd))) > 0) {
-    client->data.in += l;
+  struct evbuffer *ev = bufferevent_get_input(bev);
+  while(evbuffer_get_length(ev) >= sizeof(cmd)) {
+    evbuffer_remove(ev, &cmd, sizeof(cmd));
+    client->data.in += sizeof(cmd);
     slog(LOG_INFO, SLOG_INFO, "Read from client: %x", cmd.cmd);
     
     switch(cmd.cmd) {
       case RTL_FREQUENCY: // Frequency
-      slog(LOG_INFO, SLOG_INFO, "Set frequency: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set frequency: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_SAMPLE_RATE: // Sample rate
-      slog(LOG_INFO, SLOG_INFO, "Set sample rate: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set sample rate: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_GAIN_MODE: // Gain mode
-      slog(LOG_INFO, SLOG_INFO, "Set gain mode: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set gain mode: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_GAIN: // Set Gain
-      slog(LOG_INFO, SLOG_INFO, "Set gain: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set gain: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_FREQ_CORRECTION: // Set freq correction
-      slog(LOG_INFO, SLOG_INFO, "Set freq correction: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set freq correction: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_STAGE_GAIN: // Stage Gain
-      slog(LOG_INFO, SLOG_INFO, "Set stage gain: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set stage gain: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_TEST_MODE: // Test mode
-      slog(LOG_INFO, SLOG_INFO, "Set test mode: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set test mode: %u", ntohl(cmd.param));
       break;
       case RTL_AGC_MODE: // AGC mode
-      slog(LOG_INFO, SLOG_INFO, "Set AGC mode: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set AGC mode: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_DIRECT_SAMPLING: // Direct sampling
-      slog(LOG_INFO, SLOG_INFO, "Set direct sampling: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set direct sampling: %u", ntohl(cmd.param));
       break;
       case RTL_OFFSET_TUNING: // Offset tuning
-      slog(LOG_INFO, SLOG_INFO, "Set offset tuning: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set offset tuning: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_XTAL: // RTL Xtal
-      slog(LOG_INFO, SLOG_INFO, "Set RTL xtal: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set RTL xtal: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_TUNER_XTAL: // Tuner Xtal
-      slog(LOG_INFO, SLOG_INFO, "Set tuner xtal: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set tuner xtal: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       case RTL_GAIN_BY_INDEX: // Gain by index
-      slog(LOG_INFO, SLOG_INFO, "Set gain by index: %lu", ntohl(cmd.param));
+      slog(LOG_INFO, SLOG_INFO, "Set gain by index: %u", ntohl(cmd.param));
       serverSendCommand(cmd);
       break;
       default: // Ignore it
@@ -387,22 +432,30 @@ static void clientReadCB(struct bufferevent *bev, void *ctx) {
 
 static void connectCB(struct evconnlistener *listener,
     evutil_socket_t sock, struct sockaddr *addr, int len, void *ptr) {
+    (void)ptr;
     struct event_base *base = evconnlistener_get_base(listener);
-#ifdef THREADED
-    struct bufferevent *bev = bufferevent_socket_new(
-            base, sock, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE | BEV_OPT_DEFER_CALLBACKS);
-#else
     struct bufferevent *bev = bufferevent_socket_new(
             base, sock, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_DEFER_CALLBACKS);
-#endif
+    if(bev == NULL) {
+      slog(LOG_ERROR, SLOG_ERROR, "Could not allocate client connection.");
+      evutil_closesocket(sock);
+      return;
+    }
 
-    if (config.delayed && (serverConnection == NULL || LIST_FIRST(&clients) == NULL)) {
+    if (config.delayed && serverConnection == NULL) {
       slog(LOG_INFO, SLOG_INFO, "Connection to server triggered.");
-      connectToServer(&serverConnection);
+      connectToServer();
     }
 
-    struct client *client = addClient(bev, ptr);
-    memcpy(&client->sa, addr, len);
+    struct client *client = addClient(bev);
+    if(client == NULL) {
+      slog(LOG_ERROR, SLOG_ERROR, "Could not allocate client state.");
+      bufferevent_free(bev);
+      return;
+    }
+    if((size_t)len > sizeof(client->sin6))
+      len = sizeof(client->sin6);
+    memcpy(&client->sa, addr, (size_t)len);
     char ipBuf[128];
     if(client->sa.sa_family == AF_INET)
       evutil_inet_ntop(client->sa.sa_family, &client->sin.sin_addr, ipBuf, 128);
@@ -414,21 +467,21 @@ static void connectCB(struct evconnlistener *listener,
     bufferevent_setcb(bev, clientReadCB, NULL, errorEventCB, client);
     bufferevent_setwatermark(bev, EV_WRITE, 0, 4*1024*1024); // Limit output to 4MB
     bufferevent_enable(bev, EV_READ|EV_WRITE);
-    bufferevent_write(bev, serverInfo.magic, 4);
-    bufferevent_write(bev, &serverInfo.tuner_type, 4);
-    bufferevent_write(bev, &serverInfo.tuner_gain_count, 4);
-    serverInfo.data.out += 12;
-    client->flags = CLIENT_READY;
+    if(serverInfo.state == SERVER_CONNECTED)
+      readyClient(client);
 }
 
 static void dumpClients(struct evhttp_request *req, void *arg) {
+  (void)arg;
   struct evbuffer *evb = NULL;
 
   evb = evbuffer_new();
+  if(evb == NULL) {
+    evhttp_send_error(req, 500, "Could not allocate response");
+    return;
+  }
 
-  pthread_rwlock_rdlock(&clientLock);
-  
-  evbuffer_add_printf(evb, "{\"server\":{\"dataIn\":%lu,\"dataOut\":%lu},\"clients\":[",
+  evbuffer_add_printf(evb, "{\"server\":{\"dataIn\":%" PRIu64 ",\"dataOut\":%" PRIu64 "},\"clients\":[",
     serverInfo.data.in, serverInfo.data.out);
   struct client *client;
   LIST_FOREACH(client, &clients, peer) {
@@ -439,13 +492,13 @@ static void dumpClients(struct evhttp_request *req, void *arg) {
       evutil_inet_ntop(client->sa.sa_family, &client->sin6.sin6_addr, ipBuf, 128);
     else
       snprintf(ipBuf, 128, "from unknown address");
-    evbuffer_add_printf(evb, "{\"client\":{\"host\":\"%s\",\"port\":%u},\"dataIn\":%lu,\"dataOut\":%lu,\"dropped\":{\"size\":%lu,\"count\":%lu},\"connected\":%ld}",
+    evbuffer_add_printf(evb, "{\"client\":{\"host\":\"%s\",\"port\":%u},\"dataIn\":%" PRIu64 ",\"dataOut\":%" PRIu64 ",\"dropped\":{\"size\":%" PRIu64 ",\"count\":%" PRIu64 "},\"connected\":%lld}",
       ipBuf, ntohs(client->sa.sa_family == AF_INET ? client->sin.sin_port : client->sin6.sin6_port),
       client->data.in,
       client->data.out,
       client->data.dropped,
       client->data.droppedCount,
-      client->connected
+      (long long)client->connected
     );
     if(LIST_NEXT(client, peer) != NULL) {
       evbuffer_add_printf(evb, ",");
@@ -453,30 +506,39 @@ static void dumpClients(struct evhttp_request *req, void *arg) {
   }
   evbuffer_add_printf(evb, "]}");
   
-  pthread_rwlock_unlock(&clientLock);
-  
   evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Type", "application/json");
   evhttp_add_header(evhttp_request_get_output_headers(req), "Access-Control-Allow-Origin", "*");
   evhttp_send_reply(req, 200, "OK", evb);
+  evbuffer_free(evb);
 }
 
 void *serverThread(void *arg) {  
+  (void)arg;
+  struct evconnlistener *clientListener = NULL;
+  struct evhttp *http = NULL;
   memset(&serverInfo, 0, sizeof(serverInfo));
+  dataBlocks = 0;
+  dataBlocksSize = 0;
   
   slog(LOG_INFO, SLOG_INFO, "Starting server thread.");
   
-  LIST_INIT(&rtlDataList);
   LIST_INIT(&clients);
   
-  pthread_rwlock_init(&rtlDataLock, NULL);
-  pthread_rwlock_init(&clientLock, NULL);
-  
   event_set_log_callback(logCB);
   
-  evthread_use_pthreads();
+  if(evthread_use_pthreads() != 0) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not initialize libevent threading.");
+    timeToExit = 1;
+    return NULL;
+  }
   
   // Libevent loop
   event_base = event_base_new();
+  if(event_base == NULL) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not allocate event base.");
+    timeToExit = 1;
+    return NULL;
+  }
   
   struct sockaddr_in6 sa;
   socklen_t salen = sizeof(sa);
@@ -485,27 +547,30 @@ void *serverThread(void *arg) {
   sa.sin6_addr = in6addr_any;
   sa.sin6_port = htons(config.clientPort);
   
-  struct evconnlistener *clientListener;
   clientListener = evconnlistener_new_bind(event_base, connectCB, NULL,
     LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1,
     (struct sockaddr *)&sa, salen);
   if(!clientListener) {
     timeToExit = 1;
     slog(LOG_FATAL, SLOG_FATAL, "Could not listen on the client streaming port.");
-    return NULL;
+    goto cleanup;
   }
   
   slog(LOG_INFO, SLOG_INFO, "Listening for clients on port %d", config.clientPort);
 
   if (!config.delayed) {
-    connectToServer(&serverConnection);
+    connectToServer();
   } else {
     slog(LOG_INFO, SLOG_INFO, "Connection to server delayed.");
   }
 
-  struct evhttp *http;
   struct evhttp_bound_socket *handle;
   http = evhttp_new(event_base);
+  if(http == NULL) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not allocate HTTP server.");
+    timeToExit = 1;
+    goto cleanup;
+  }
 
   evhttp_set_cb(http, "/stats.json", dumpClients, "clients");
   
@@ -514,7 +579,7 @@ void *serverThread(void *arg) {
   if(!handle) {
     slog(LOG_FATAL, SLOG_FATAL, "Could not bind HTTP listener.");
     timeToExit = 1;
-    return NULL;
+    goto cleanup;
   }
 
   int loopCounter = 0;
@@ -528,42 +593,34 @@ void *serverThread(void *arg) {
     
     if((++loopCounter%600) == 0) {
       loopCounter = 0;
-/*      pthread_rwlock_rdlock(&clientLock);
-      struct client *client;
-      unsigned long clientCount = 0;
-      LIST_FOREACH(client, &clients, peer) {
-        clientCount++;
-      }
-      slog(LOG_INFO, SLOG_INFO, "Clients currently connected: %lu", clientCount);
-      pthread_rwlock_unlock(&clientLock);*/
-      pthread_rwlock_rdlock(&rtlDataLock);
       if(dataBlocks > 0)
         slog(LOG_INFO, SLOG_INFO, "Maintaining %lu data buffers, total of %lu bytes.", dataBlocks, dataBlocksSize);
-      pthread_rwlock_unlock(&rtlDataLock);
     }
   }
   
-  pthread_rwlock_wrlock(&clientLock);
+cleanup:
   while(LIST_FIRST(&clients) != NULL) {
     struct client *client = LIST_FIRST(&clients);
     LIST_REMOVE(client, peer);
     bufferevent_free(client->bev);
     free(client);
   }
-  pthread_rwlock_unlock(&clientLock);
-  
-  evconnlistener_free(clientListener);
-  evhttp_free(http);
-  
-  event_base_free(event_base);
 
   if (serverConnection != NULL) {
-    bufferevent_free(serverConnection);
-    serverConnection = NULL;
+    disconnectServer(serverConnection);
     serverInfo.state = SERVER_DISCONNECTED;
     slog(LOG_INFO, SLOG_INFO, "Disconnecting from server.");
   }
 
+  if(clientListener != NULL)
+    evconnlistener_free(clientListener);
+  if(http != NULL)
+    evhttp_free(http);
+  if(event_base != NULL) {
+    event_base_free(event_base);
+    event_base = NULL;
+  }
+
   slog(LOG_INFO, SLOG_INFO, "End of server thread.");
   return NULL;
 }
diff --git a/rtlmux.h b/rtlmux.h
index b1e2789..e7620f7 100644
--- a/rtlmux.h
+++ b/rtlmux.h
@@ -1,13 +1,16 @@
 #ifndef _SERVER_H_
 #define _SERVER_H_
 
-extern volatile unsigned char timeToExit;
+#include <signal.h>
+#include <stdint.h>
+
+extern volatile sig_atomic_t timeToExit;
 
 extern void *serverThread(void *);
 
 struct command {
-        unsigned char cmd;
-        unsigned int param;
+        uint8_t cmd;
+        uint32_t param;
 }__attribute__((packed));
 
 #endif
diff --git a/slog.c b/slog.c
index 6bf802d..815d5b4 100644
--- a/slog.c
+++ b/slog.c
@@ -301,7 +301,7 @@ void slog(int level, int flag, const char *msg, ...)
     /* Read args */
     va_list args;
     va_start(args, msg);
-    vsprintf(string, msg, args);
+    vsnprintf(string, sizeof(string), msg, args);
     va_end(args);
 
     /* Check logging levels */
@@ -350,7 +350,7 @@ void slog(int level, int flag, const char *msg, ...)
         /* Print output */
         if (level <= slg.level || slg.pretty)
         {
-            if (flag != SLOG_NONE) sprintf(prints, "[%s] %s", strclr(color, alarm), string);
+            if (flag != SLOG_NONE) snprintf(prints, sizeof(prints), "[%s] %s", strclr(color, alarm), string);
             if (level <= slg.level) printf("%s", slog_get(&mdate, "%s\n", prints));
         }
 
@@ -360,7 +360,7 @@ void slog(int level, int flag, const char *msg, ...)
             if (slg.pretty) output = slog_get(&mdate, "%s\n", prints);
             else 
             {
-                if (flag != SLOG_NONE) sprintf(prints, "[%s] %s", alarm, string);
+                if (flag != SLOG_NONE) snprintf(prints, sizeof(prints), "[%s] %s", alarm, string);
                 output = slog_get(&mdate, "%s\n", prints);
             } 
 
diff --git a/tests/integration.py b/tests/integration.py
new file mode 100644
index 0000000..8383222
--- /dev/null
+++ b/tests/integration.py
@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+
+import socket
+import subprocess
+import sys
+import threading
+import time
+import unittest
+
+
+BINARY = sys.argv[1] if len(sys.argv) > 1 else "./rtlmux"
+sys.argv = [sys.argv[0]]
+
+
+def unused_port_pair():
+    while True:
+        first = socket.socket()
+        first.bind(("127.0.0.1", 0))
+        port = first.getsockname()[1]
+        second = socket.socket()
+        try:
+            second.bind(("127.0.0.1", port + 1))
+        except OSError:
+            first.close()
+            second.close()
+            continue
+        first.close()
+        second.close()
+        return port
+
+
+def connect_with_retry(port):
+    deadline = time.time() + 8
+    while time.time() < deadline:
+        try:
+            return socket.create_connection(("127.0.0.1", port), timeout=1)
+        except OSError:
+            time.sleep(0.03)
+    raise RuntimeError("rtlmux did not start listening")
+
+
+class RtlmuxTest(unittest.TestCase):
+    def start_rtlmux(self, upstream_port, listen_port, *args):
+        process = subprocess.Popen(
+            [
+                BINARY,
+                "-h", "127.0.0.1",
+                "-p", str(upstream_port),
+                "-l", str(listen_port),
+                *args,
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        self.addCleanup(self.stop_process, process)
+        return process
+
+    @staticmethod
+    def stop_process(process):
+        if process.poll() is not None:
+            return
+        process.terminate()
+        try:
+            process.wait(timeout=3)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait()
+
+    def test_delayed_client_waits_for_real_header(self):
+        upstream = socket.socket()
+        upstream.bind(("127.0.0.1", 0))
+        upstream.listen()
+        upstream_port = upstream.getsockname()[1]
+        listen_port = unused_port_pair()
+        accepted = threading.Event()
+        release_header = threading.Event()
+
+        def serve():
+            connection, _ = upstream.accept()
+            accepted.set()
+            release_header.wait(2)
+            connection.sendall(b"RTL0" + (5).to_bytes(4, "big") + (7).to_bytes(4, "big"))
+            time.sleep(0.2)
+            connection.close()
+            upstream.close()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        self.addCleanup(thread.join, 2)
+        self.start_rtlmux(upstream_port, listen_port, "-d")
+
+        client = connect_with_retry(listen_port)
+        self.addCleanup(client.close)
+        self.assertTrue(accepted.wait(2))
+        client.settimeout(0.15)
+        with self.assertRaises(socket.timeout):
+            client.recv(12)
+
+        release_header.set()
+        client.settimeout(2)
+        self.assertEqual(client.recv(12), b"RTL0" + (5).to_bytes(4, "big") + (7).to_bytes(4, "big"))
+
+    def test_fragmented_command_is_forwarded_when_complete(self):
+        upstream = socket.socket()
+        upstream.bind(("127.0.0.1", 0))
+        upstream.listen()
+        upstream_port = upstream.getsockname()[1]
+        listen_port = unused_port_pair()
+        received = bytearray()
+        ready = threading.Event()
+        partial_checked = threading.Event()
+        finish = threading.Event()
+
+        def serve():
+            connection, _ = upstream.accept()
+            connection.sendall(b"RTL0" + (1).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+            connection.settimeout(2)
+            ready.set()
+            try:
+                received.extend(connection.recv(5))
+            except socket.timeout:
+                pass
+            partial_checked.set()
+            connection.settimeout(2)
+            while len(received) < 5:
+                try:
+                    chunk = connection.recv(5 - len(received))
+                except socket.timeout:
+                    break
+                if not chunk:
+                    break
+                received.extend(chunk)
+            finish.set()
+            connection.close()
+            upstream.close()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        self.addCleanup(thread.join, 2)
+        self.start_rtlmux(upstream_port, listen_port)
+        self.assertTrue(ready.wait(2))
+
+        client = connect_with_retry(listen_port)
+        self.addCleanup(client.close)
+        client.settimeout(2)
+        self.assertEqual(client.recv(12), b"RTL0" + (1).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+
+        command = b"\x01" + (100_000_000).to_bytes(4, "big")
+        client.sendall(command[:1])
+        self.assertTrue(partial_checked.wait(4))
+        self.assertEqual(received, b"")
+        client.sendall(command[1:])
+
+        self.assertTrue(finish.wait(2))
+        self.assertEqual(received, command)
+
+    def test_invalid_ports_are_rejected(self):
+        upstream_result = subprocess.run(
+            [BINARY, "-p", "-1"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=2,
+        )
+        self.assertNotEqual(upstream_result.returncode, 0)
+        self.assertIn("rtl_tcp port must be between", upstream_result.stderr)
+
+        listen_result = subprocess.run(
+            [BINARY, "-l", "70000"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=2,
+        )
+        self.assertNotEqual(listen_result.returncode, 0)
+        self.assertIn("Listening port must be between", listen_result.stderr)
+
+    def test_shutdown_during_reconnect_is_clean(self):
+        upstream = socket.socket()
+        upstream.bind(("127.0.0.1", 0))
+        upstream.listen()
+        upstream_port = upstream.getsockname()[1]
+        listen_port = unused_port_pair()
+        closed = threading.Event()
+
+        def serve():
+            connection, _ = upstream.accept()
+            connection.sendall(b"RTL0" + (1).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+            connection.close()
+            upstream.close()
+            closed.set()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        self.addCleanup(thread.join, 2)
+        process = self.start_rtlmux(upstream_port, listen_port)
+
+        self.assertTrue(closed.wait(2))
+        time.sleep(0.2)
+        process.terminate()
+        self.assertEqual(process.wait(timeout=3), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

From 89230c7f094064bac25166d757e5f90ec43d6c07 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:36:07 -0600
Subject: [PATCH 2/8] build: modernize portable static builds

---
 .dockerignore |  5 ++++
 .gitignore    |  1 +
 Dockerfile    | 31 +++++++++++++++++------
 Makefile      | 70 ++++++++++++++++++++++++++++++---------------------
 4 files changed, 72 insertions(+), 35 deletions(-)
 create mode 100644 .dockerignore

diff --git a/.dockerignore b/.dockerignore
new file mode 100644
index 0000000..be6fed4
--- /dev/null
+++ b/.dockerignore
@@ -0,0 +1,5 @@
+.git
+.github
+*.d
+*.o
+rtlmux
diff --git a/.gitignore b/.gitignore
index f894974..03f0f67 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
 /*.o
+/*.d
 /rtlmux
 _codeql_detected_source_root
diff --git a/Dockerfile b/Dockerfile
index e93a688..3ec11f1 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,13 +1,30 @@
-FROM alpine
+# syntax=docker/dockerfile:1
 
-RUN apk --no-cache add libevent
+FROM alpine:3.22 AS build
 
-COPY Makefile *.c *.h options.ggo /app/
+ARG TARGETARCH
+ARG TARGETVARIANT
+
+RUN apk --no-cache add bsd-compat-headers build-base libevent-dev libevent-static pkgconf
 
 WORKDIR /app
+COPY Makefile *.c *.h options.ggo ./
+COPY tests tests
+
+RUN make static \
+  && strip rtlmux \
+  && case "${TARGETARCH}/${TARGETVARIANT}" in \
+       amd64/) name=amd64 ;; \
+       arm64/) name=arm64 ;; \
+       arm/v7) name=armv7 ;; \
+       *) echo "Unsupported target: ${TARGETARCH}/${TARGETVARIANT}" >&2; exit 1 ;; \
+     esac \
+  && mkdir /out \
+  && cp rtlmux "/out/rtlmux-linux-${name}"
 
-RUN apk --no-cache add --virtual build-dependencies build-base libevent-dev bsd-compat-headers \
-  && make \
-  && apk del build-dependencies
+FROM scratch AS release
+COPY --from=build /out/ /
 
-CMD ["/app/rtlmux"]
+FROM scratch AS runtime
+COPY --from=build /app/rtlmux /rtlmux
+ENTRYPOINT ["/rtlmux"]
diff --git a/Makefile b/Makefile
index 4d91a48..aa9f605 100644
--- a/Makefile
+++ b/Makefile
@@ -1,41 +1,55 @@
-CC=gcc
-CFLAGS=-Wall -I/usr/local/include -O3
-
-LIBS=-pthread `pkg-config libevent --libs-only-L` -levent -levent_pthreads
-
-ifeq ($(shell uname -m),armv7l)
-  CFLAGS+=-O3 -mfpu=neon-vfpv4 -mfloat-abi=hard -march=armv7-a -ffast-math -funsafe-math-optimizations
-else
-  CFLAGS+=-O3
+CC ?= cc
+PKG_CONFIG ?= pkg-config
+PYTHON ?= python3
+GENGETOPT ?= gengetopt
+
+PREFIX ?= /usr/local
+DESTDIR ?=
+
+CPPFLAGS += $(shell $(PKG_CONFIG) --cflags libevent_pthreads)
+CFLAGS ?= -O2 -g
+CFLAGS += -Wall -Wextra -Wformat=2 -pthread
+LDFLAGS ?=
+LDLIBS += -pthread $(shell $(PKG_CONFIG) --libs libevent_pthreads)
+
+ifeq ($(STATIC),1)
+  LDFLAGS += -static
+  LDLIBS := -pthread $(shell $(PKG_CONFIG) --static --libs libevent_pthreads)
 endif
 
-ifneq ($(shell uname),Darwin)
-  LIBS+=-lrt
-endif
+SRC = slog.c rtlmux.c config.c cmdline.c main.c
+OBJS = $(SRC:.c=.o)
+DEPS = $(OBJS:.o=.d)
 
-ifeq ($(shell uname),Darwin)
-  CFLAGS+=-glldb
-else
-  CFLAGS+=-ggdb3
-endif
+.PHONY: all clean install static test
 
-SRC=slog.c rtlmux.c config.c cmdline.c main.c
-OBJS=slog.o rtlmux.o config.o cmdline.o main.o
+all: rtlmux
 
-all: cmdline.c rtlmux
+rtlmux: $(OBJS)
+	$(CC) $(LDFLAGS) -o $@ $^ $(LDLIBS)
 
-clean:
-	rm -f rtlmux $(OBJS)
+%.o: %.c
+	$(CC) $(CPPFLAGS) $(CFLAGS) -MMD -MP -c -o $@ $<
 
-depend:
-	makedepend -- $(CFLAGS) -- $(SRC)
+cmdline.o: CFLAGS += -Wno-unused-but-set-variable
 
 cmdline.h: options.ggo
-	gengetopt -C -i $< -f cmdline.c
+	$(GENGETOPT) -C -i $< -f cmdline.c
 
 cmdline.c: cmdline.h
 
-rtlmux: $(OBJS)
-	$(CC) $(CFLAGS) -o $@ $^ $(LIBS)
+static:
+	$(MAKE) clean
+	$(MAKE) STATIC=1 all
+
+test: rtlmux
+	$(PYTHON) tests/integration.py ./rtlmux
+
+install: rtlmux
+	install -d $(DESTDIR)$(PREFIX)/bin
+	install -m 0755 rtlmux $(DESTDIR)$(PREFIX)/bin/rtlmux
+
+clean:
+	rm -f rtlmux $(OBJS) $(DEPS)
 
-# DO NOT DELETE
+-include $(DEPS)

From 12e8f902b5effa9b8315a19469fc35148eec4c78 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:36:33 -0600
Subject: [PATCH 3/8] ci: publish binaries and Docker images

---
 .github/workflows/ci.yml      |  39 +++++++++++
 .github/workflows/release.yml | 127 ++++++++++++++++++++++++++++++++++
 2 files changed, 166 insertions(+)
 create mode 100644 .github/workflows/ci.yml
 create mode 100644 .github/workflows/release.yml

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..9dad0f9
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,39 @@
+name: CI
+
+on:
+  push:
+    branches:
+      - main
+      - master
+  pull_request:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Install build dependencies
+        run: sudo apt-get update && sudo apt-get install --yes libevent-dev
+
+      - name: Build and test
+        run: make test
+
+      - uses: docker/setup-buildx-action@v4
+
+      - name: Build static Linux binary
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          platforms: linux/amd64
+          target: release
+          outputs: type=local,dest=dist
+
+      - name: Verify static Linux binary
+        run: |
+          file dist/rtlmux-linux-amd64
+          file dist/rtlmux-linux-amd64 | grep -q "statically linked"
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
new file mode 100644
index 0000000..a3b5a6c
--- /dev/null
+++ b/.github/workflows/release.yml
@@ -0,0 +1,127 @@
+name: Release
+
+on:
+  push:
+    branches:
+      - main
+      - master
+    tags:
+      - "v*"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  static-binaries:
+    if: startsWith(github.ref, 'refs/tags/v')
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+          - platform: linux/amd64
+            name: amd64
+          - platform: linux/arm64
+            name: arm64
+          - platform: linux/arm/v7
+            name: armv7
+    steps:
+      - uses: actions/checkout@v7
+
+      - uses: docker/setup-qemu-action@v4
+
+      - uses: docker/setup-buildx-action@v4
+
+      - name: Build static binary
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          platforms: ${{ matrix.platform }}
+          target: release
+          outputs: type=local,dest=dist
+
+      - name: Verify artifact
+        run: |
+          test -x "dist/rtlmux-linux-${{ matrix.name }}"
+          file "dist/rtlmux-linux-${{ matrix.name }}" | grep -q "statically linked"
+
+      - uses: actions/upload-artifact@v7
+        with:
+          name: rtlmux-linux-${{ matrix.name }}
+          path: dist/rtlmux-linux-${{ matrix.name }}
+          if-no-files-found: error
+
+  github-release:
+    if: startsWith(github.ref, 'refs/tags/v')
+    needs: static-binaries
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+    steps:
+      - uses: actions/download-artifact@v8
+        with:
+          path: dist
+          merge-multiple: true
+
+      - name: Create checksums
+        run: |
+          cd dist
+          sha256sum rtlmux-linux-* > SHA256SUMS
+
+      - name: Publish GitHub release
+        env:
+          GH_TOKEN: ${{ github.token }}
+          GH_REPO: ${{ github.repository }}
+        run: gh release create "$GITHUB_REF_NAME" dist/* --verify-tag --generate-notes
+
+  docker:
+    runs-on: ubuntu-latest
+    env:
+      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+      DOCKERHUB_TOKEN: ${{ secrets.DOCKERHUB_TOKEN }}
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Check Docker Hub credentials
+        id: dockerhub
+        run: |
+          if [ -n "$DOCKERHUB_USERNAME" ] && [ -n "$DOCKERHUB_TOKEN" ]; then
+            echo "enabled=true" >> "$GITHUB_OUTPUT"
+          else
+            echo "::notice::Docker Hub publishing skipped; configure DOCKERHUB_USERNAME and DOCKERHUB_TOKEN."
+            echo "enabled=false" >> "$GITHUB_OUTPUT"
+          fi
+
+      - if: steps.dockerhub.outputs.enabled == 'true'
+        uses: docker/setup-qemu-action@v4
+
+      - if: steps.dockerhub.outputs.enabled == 'true'
+        uses: docker/setup-buildx-action@v4
+
+      - if: steps.dockerhub.outputs.enabled == 'true'
+        uses: docker/login-action@v4
+        with:
+          username: ${{ env.DOCKERHUB_USERNAME }}
+          password: ${{ env.DOCKERHUB_TOKEN }}
+
+      - if: steps.dockerhub.outputs.enabled == 'true'
+        id: metadata
+        uses: docker/metadata-action@v6
+        with:
+          images: slepp/rtlmux
+          tags: |
+            type=raw,value=edge,enable=${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/master' }}
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/v') }}
+
+      - name: Publish Docker image
+        if: steps.dockerhub.outputs.enabled == 'true'
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64,linux/arm/v7
+          target: runtime
+          push: true
+          tags: ${{ steps.metadata.outputs.tags }}
+          labels: ${{ steps.metadata.outputs.labels }}

From e3b2d008700f6b8d0806b799bc33cfdacc45cac7 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:36:40 -0600
Subject: [PATCH 4/8] docs: refresh setup and release guidance

---
 .gitlab-ci.yml |   9 -----
 README.md      | 102 +++++++++++++++++++++++++++++++++++++++----------
 2 files changed, 82 insertions(+), 29 deletions(-)
 delete mode 100644 .gitlab-ci.yml

diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
deleted file mode 100644
index a2b22cf..0000000
--- a/.gitlab-ci.yml
+++ /dev/null
@@ -1,9 +0,0 @@
-stages:
-- build
-
-build:make:
- image: alpine
- stage: build
- script:
- - "apk --update add --virtual build-dependencies build-base libevent-dev bsd-compat-headers"
- - make
diff --git a/README.md b/README.md
index c84925d..25ae3a5 100644
--- a/README.md
+++ b/README.md
@@ -1,27 +1,89 @@
-RTL TCP Client Multiplexer / Relay
-==================================
+# RTL TCP Client Multiplexer / Relay
 
-[![Build Status](https://git.vocti.ca/slepp/rtlmux/badges/master/build.svg)](https://git.vocti.ca/slepp/rtlmux/builds)
+`rtlmux` connects to one [`rtl_tcp`](https://osmocom.org/projects/rtl-sdr/wiki/Rtl-sdr) server and lets multiple network clients share its sample stream. Clients receive the normal RTL TCP header and sample data, and supported tuner commands are forwarded to the upstream receiver.
 
-This is a simple server to connec to an rtl_tcp server and allow multiple clients
-to attach to the same receiver. It relays the ID information to the clients,
-and allows all clients to send commands for configuration to the rtl_tcp server.
+The service also provides a small JSON status endpoint with connection and traffic statistics.
 
-It provides status output of the server and clients on the client port + 1, at
-the URL /stats.json, ie. http://localhost:7879/stats.json when `-l` is set to 7878.
+> [!IMPORTANT]
+> RTL TCP has no authentication or encryption, and every connected client can change the shared tuner configuration. Run `rtlmux` only on a trusted LAN or behind an appropriate firewall or VPN.
 
-Usage
-=====
+## Quick start
 
-Usage takes up to 3 command line options:
+Download the static binary for your Linux system from the [latest GitHub release](https://github.com/slepp/rtlmux/releases/latest):
 
-* `-h host` Server host to connect to
-* `-p port` Server port to connect to
-* `-l port` Listening port for clients to connect to, HTTP status port will be
-  equal to this plus one
-* `-d`      Connect to server only when first client arrives and exit when last disconnects
-* `-r`      Close connection with the server and restart it when last client disconnects
+```sh
+# x86_64 / amd64
+curl -LO https://github.com/slepp/rtlmux/releases/latest/download/rtlmux-linux-amd64
+chmod +x rtlmux-linux-amd64
 
-Note: For standby use (for example, when running as a daemon), use both flags `-d -r`.
-When running in this way the RTL_TCP is only active (reading samples) when some client is connected.
-In any other case, the RTL_TCP and RTLMUX processes are in standby (idle).
+# Connect to rtl_tcp at 192.168.1.50:1234 and listen for clients on port 7878
+./rtlmux-linux-amd64 -h 192.168.1.50 -p 1234 -l 7878
+```
+
+Release binaries are available for:
+
+| File | Platform |
+| --- | --- |
+| `rtlmux-linux-amd64` | 64-bit Intel and AMD Linux |
+| `rtlmux-linux-arm64` | 64-bit ARM Linux |
+| `rtlmux-linux-armv7` | 32-bit ARMv7 Linux, including WEB-888 |
+
+Point RTL TCP clients at the machine running `rtlmux`, port `7878` in the example above. Status is available at <http://localhost:7879/stats.json>.
+
+## Usage
+
+```text
+rtlmux [OPTIONS]
+
+  -h, --host=ADDRESS  rtl_tcp server address (default: localhost)
+  -p, --port=PORT     rtl_tcp server port (default: 1234)
+  -l, --listen=PORT   client listening port (default: 7878)
+  -d, --delayed       connect upstream only when a client arrives
+  -r, --restart       restart after the last client disconnects
+  -V, --version       print version and exit
+      --help          print complete help and exit
+```
+
+The HTTP status endpoint uses the port immediately after the client port. For example, `-l 7878` uses port `7879` for HTTP.
+
+For standby operation, use `-d -r` together:
+
+```sh
+./rtlmux-linux-armv7 -h 192.168.1.50 -p 1234 -l 7878 -d -r
+```
+
+This keeps the upstream connection idle until the first client arrives, then closes and resets it after the last client disconnects.
+
+## Docker
+
+Multi-architecture images are published to [Docker Hub](https://hub.docker.com/r/slepp/rtlmux) for amd64, arm64, and ARMv7:
+
+```sh
+docker run --rm --network host slepp/rtlmux:latest \
+  -h 192.168.1.50 -p 1234 -l 7878 -d -r
+```
+
+Version tags and `latest` are published from Git version tags. The `edge` image follows the repository's primary branch.
+
+## Build from source
+
+Install a C compiler, `make`, `pkg-config`, the libevent development files, and Python 3. On Debian or Ubuntu:
+
+```sh
+sudo apt-get install build-essential libevent-dev pkg-config python3
+make
+make test
+sudo make install
+```
+
+`make static` creates a statically linked binary when static libevent libraries are installed. The Docker build provides a reproducible Alpine/musl static build:
+
+```sh
+docker build --target runtime -t rtlmux .
+```
+
+## Releases
+
+Pushing a tag such as `v1.1.0` builds static Linux binaries for amd64, arm64, and ARMv7, creates checksums, and publishes a GitHub release.
+
+Docker Hub publishing uses the repository secrets `DOCKERHUB_USERNAME` and `DOCKERHUB_TOKEN`. Primary-branch builds publish `edge`; version tags publish the version and `latest`.

From 79234fbf5c95b60116cf69bd80580c37e0265e43 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:40:40 -0600
Subject: [PATCH 5/8] fix: align release and standby behaviour

---
 .github/workflows/release.yml | 7 +++++--
 README.md                     | 4 ++--
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index a3b5a6c..6c46da6 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -85,10 +85,13 @@ jobs:
       - name: Check Docker Hub credentials
         id: dockerhub
         run: |
-          if [ -n "$DOCKERHUB_USERNAME" ] && [ -n "$DOCKERHUB_TOKEN" ]; then
+          if [ -z "$DOCKERHUB_USERNAME" ] || [ -z "$DOCKERHUB_TOKEN" ]; then
+            echo "::notice::Docker Hub publishing skipped; configure DOCKERHUB_USERNAME and DOCKERHUB_TOKEN."
+            echo "enabled=false" >> "$GITHUB_OUTPUT"
+          elif [[ "$GITHUB_REF" == "refs/heads/main" || "$GITHUB_REF" == "refs/heads/master" || "$GITHUB_REF" == refs/tags/v* ]]; then
             echo "enabled=true" >> "$GITHUB_OUTPUT"
           else
-            echo "::notice::Docker Hub publishing skipped; configure DOCKERHUB_USERNAME and DOCKERHUB_TOKEN."
+            echo "::notice::Docker Hub publishing skipped for $GITHUB_REF."
             echo "enabled=false" >> "$GITHUB_OUTPUT"
           fi
 
diff --git a/README.md b/README.md
index 25ae3a5..58bf312 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@ rtlmux [OPTIONS]
   -h, --host=ADDRESS  rtl_tcp server address (default: localhost)
   -p, --port=PORT     rtl_tcp server port (default: 1234)
   -l, --listen=PORT   client listening port (default: 7878)
-  -d, --delayed       connect upstream only when a client arrives
+  -d, --delayed       connect upstream on demand and exit after the last client
   -r, --restart       restart after the last client disconnects
   -V, --version       print version and exit
       --help          print complete help and exit
@@ -46,7 +46,7 @@ rtlmux [OPTIONS]
 
 The HTTP status endpoint uses the port immediately after the client port. For example, `-l 7878` uses port `7879` for HTTP.
 
-For standby operation, use `-d -r` together:
+With `-d` alone, `rtlmux` exits after the last client disconnects. For a long-running standby service, use `-d -r` together:
 
 ```sh
 ./rtlmux-linux-armv7 -h 192.168.1.50 -p 1234 -l 7878 -d -r

From 727ae749208d47601fe9f72e1f2b24043450ce23 Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:42:19 -0600
Subject: [PATCH 6/8] test: cover standby restart behaviour

---
 tests/integration.py | 39 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 39 insertions(+)

diff --git a/tests/integration.py b/tests/integration.py
index 8383222..8771007 100644
--- a/tests/integration.py
+++ b/tests/integration.py
@@ -200,6 +200,45 @@ def serve():
         process.terminate()
         self.assertEqual(process.wait(timeout=3), 0)
 
+    def test_delayed_restart_accepts_a_new_client(self):
+        upstream = socket.socket()
+        upstream.bind(("127.0.0.1", 0))
+        upstream.listen()
+        upstream_port = upstream.getsockname()[1]
+        listen_port = unused_port_pair()
+        disconnected = [threading.Event(), threading.Event()]
+
+        def serve():
+            for index in range(2):
+                connection, _ = upstream.accept()
+                connection.sendall(b"RTL0" + (index + 1).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+                connection.settimeout(3)
+                try:
+                    while connection.recv(1024):
+                        pass
+                except socket.timeout:
+                    pass
+                connection.close()
+                disconnected[index].set()
+            upstream.close()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        self.addCleanup(thread.join, 4)
+        self.start_rtlmux(upstream_port, listen_port, "-d", "-r")
+
+        first = connect_with_retry(listen_port)
+        first.settimeout(2)
+        self.assertEqual(first.recv(12), b"RTL0" + (1).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+        first.close()
+        self.assertTrue(disconnected[0].wait(3))
+
+        second = connect_with_retry(listen_port)
+        second.settimeout(2)
+        self.assertEqual(second.recv(12), b"RTL0" + (2).to_bytes(4, "big") + (2).to_bytes(4, "big"))
+        second.close()
+        self.assertTrue(disconnected[1].wait(3))
+
 
 if __name__ == "__main__":
     unittest.main()

From 73b86dfef4a4ef7efc3dff490c1ee47744d649af Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 17:42:19 -0600
Subject: [PATCH 7/8] fix: make parser generation explicit

---
 Makefile  | 6 ++----
 README.md | 2 ++
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/Makefile b/Makefile
index aa9f605..2b42edb 100644
--- a/Makefile
+++ b/Makefile
@@ -21,7 +21,7 @@ SRC = slog.c rtlmux.c config.c cmdline.c main.c
 OBJS = $(SRC:.c=.o)
 DEPS = $(OBJS:.o=.d)
 
-.PHONY: all clean install static test
+.PHONY: all clean generate install static test
 
 all: rtlmux
 
@@ -33,11 +33,9 @@ rtlmux: $(OBJS)
 
 cmdline.o: CFLAGS += -Wno-unused-but-set-variable
 
-cmdline.h: options.ggo
+generate: options.ggo
 	$(GENGETOPT) -C -i $< -f cmdline.c
 
-cmdline.c: cmdline.h
-
 static:
 	$(MAKE) clean
 	$(MAKE) STATIC=1 all
diff --git a/README.md b/README.md
index 58bf312..483c007 100644
--- a/README.md
+++ b/README.md
@@ -82,6 +82,8 @@ sudo make install
 docker build --target runtime -t rtlmux .
 ```
 
+The generated command-line parser is committed to the repository. Install `gengetopt` and run `make generate` only when changing `options.ggo`.
+
 ## Releases
 
 Pushing a tag such as `v1.1.0` builds static Linux binaries for amd64, arm64, and ARMv7, creates checksums, and publishes a GitHub release.

From e0e35f9abdc2d23bcc725936e306b7fe2e3db96b Mon Sep 17 00:00:00 2001
From: Stephen Olesen <slepp@slepp.ca>
Date: Sun, 2 Aug 2026 18:09:24 -0600
Subject: [PATCH 8/8] fix: address pull request review

---
 main.c               | 19 +------------------
 rtlmux.c             | 24 ++++++++++++++++++++++++
 rtlmux.h             |  3 +--
 tests/integration.py | 24 ++++++++++++------------
 4 files changed, 38 insertions(+), 32 deletions(-)

diff --git a/main.c b/main.c
index 6788ab8..0c4db0c 100644
--- a/main.c
+++ b/main.c
@@ -3,17 +3,10 @@
 
 #include "rtlmux.h"
 
-#include <errno.h>
 #include <pthread.h>
-#include <signal.h>
 #include <string.h>
 
-volatile sig_atomic_t timeToExit = 0;
-
-void signalExit(int sig) {
-  (void)sig;
-  timeToExit = 1;
-}
+unsigned char timeToExit = 0;
 
 int main(int argc, char **argv) {
   pthread_t threadServer;
@@ -21,16 +14,6 @@ int main(int argc, char **argv) {
   
   parseConfig(argc, argv);
   slog_init(NULL, NULL, LOG_EXTRA, LOG_DEBUG, 1);
-
-  struct sigaction sigact;
-  memset(&sigact, 0, sizeof(sigact));
-  sigact.sa_handler = signalExit;
-  sigact.sa_flags = 0;
-  sigemptyset(&sigact.sa_mask);
-  if(sigaction(SIGTERM, &sigact, NULL) != 0 || sigaction(SIGINT, &sigact, NULL) != 0) {
-    slog(LOG_FATAL, SLOG_FATAL, "Could not configure signal handlers: %s", strerror(errno));
-    return 1;
-  }
   
   do {
     result = pthread_create(&threadServer, NULL, serverThread, NULL);
diff --git a/rtlmux.c b/rtlmux.c
index 820c7ef..0170f3e 100644
--- a/rtlmux.c
+++ b/rtlmux.c
@@ -5,6 +5,7 @@
 #include <unistd.h>
 #include <string.h>
 #include <errno.h>
+#include <signal.h>
 #include <time.h>
 #include <stdarg.h>
 #include <math.h>
@@ -175,6 +176,13 @@ static void readyWaitingClients(void) {
   }
 }
 
+static void signalEventCB(evutil_socket_t signal, short events, void *ctx) {
+  (void)events;
+  (void)ctx;
+  slog(LOG_INFO, SLOG_INFO, "Received signal %d.", (int)signal);
+  timeToExit = 1;
+}
+
 static void serverErrorEventCB(struct bufferevent *, short, void *);
 static void serverReadCB(struct bufferevent *, void *);
 static void connectToServerSoon(void);
@@ -245,6 +253,7 @@ static void serverReadCB(struct bufferevent *bev, void *ctx) {
     serverInfo.data.in += evbuffer_remove(ev, &serverInfo.tuner_gain_count, 4);
     if(serverInfo.magic[0] == 'R' && serverInfo.magic[1] == 'T' && serverInfo.magic[2] == 'L' && serverInfo.magic[3] == '0') {
       serverInfo.state = SERVER_CONNECTED;
+      bufferevent_setwatermark(bev, EV_READ, 16384, 0);
       slog(LOG_INFO, SLOG_INFO, "Connected to server.");
       readyWaitingClients();
     } else { // Failed to receive the magic header
@@ -516,6 +525,8 @@ void *serverThread(void *arg) {
   (void)arg;
   struct evconnlistener *clientListener = NULL;
   struct evhttp *http = NULL;
+  struct event *sigtermEvent = NULL;
+  struct event *sigintEvent = NULL;
   memset(&serverInfo, 0, sizeof(serverInfo));
   dataBlocks = 0;
   dataBlocksSize = 0;
@@ -539,6 +550,15 @@ void *serverThread(void *arg) {
     timeToExit = 1;
     return NULL;
   }
+
+  sigtermEvent = evsignal_new(event_base, SIGTERM, signalEventCB, NULL);
+  sigintEvent = evsignal_new(event_base, SIGINT, signalEventCB, NULL);
+  if(sigtermEvent == NULL || sigintEvent == NULL ||
+      event_add(sigtermEvent, NULL) != 0 || event_add(sigintEvent, NULL) != 0) {
+    slog(LOG_FATAL, SLOG_FATAL, "Could not configure signal events.");
+    timeToExit = 1;
+    goto cleanup;
+  }
   
   struct sockaddr_in6 sa;
   socklen_t salen = sizeof(sa);
@@ -616,6 +636,10 @@ void *serverThread(void *arg) {
     evconnlistener_free(clientListener);
   if(http != NULL)
     evhttp_free(http);
+  if(sigtermEvent != NULL)
+    event_free(sigtermEvent);
+  if(sigintEvent != NULL)
+    event_free(sigintEvent);
   if(event_base != NULL) {
     event_base_free(event_base);
     event_base = NULL;
diff --git a/rtlmux.h b/rtlmux.h
index e7620f7..e2aa86a 100644
--- a/rtlmux.h
+++ b/rtlmux.h
@@ -1,10 +1,9 @@
 #ifndef _SERVER_H_
 #define _SERVER_H_
 
-#include <signal.h>
 #include <stdint.h>
 
-extern volatile sig_atomic_t timeToExit;
+extern unsigned char timeToExit;
 
 extern void *serverThread(void *);
 
diff --git a/tests/integration.py b/tests/integration.py
index 8771007..6c996e4 100644
--- a/tests/integration.py
+++ b/tests/integration.py
@@ -40,6 +40,14 @@ def connect_with_retry(port):
 
 
 class RtlmuxTest(unittest.TestCase):
+    def start_server_thread(self, listener, target, join_timeout=4):
+        listener.settimeout(join_timeout)
+        thread = threading.Thread(target=target, daemon=True)
+        thread.start()
+        self.addCleanup(thread.join, join_timeout)
+        self.addCleanup(listener.close)
+        return thread
+
     def start_rtlmux(self, upstream_port, listen_port, *args):
         process = subprocess.Popen(
             [
@@ -84,9 +92,7 @@ def serve():
             connection.close()
             upstream.close()
 
-        thread = threading.Thread(target=serve)
-        thread.start()
-        self.addCleanup(thread.join, 2)
+        self.start_server_thread(upstream, serve)
         self.start_rtlmux(upstream_port, listen_port, "-d")
 
         client = connect_with_retry(listen_port)
@@ -134,9 +140,7 @@ def serve():
             connection.close()
             upstream.close()
 
-        thread = threading.Thread(target=serve)
-        thread.start()
-        self.addCleanup(thread.join, 2)
+        self.start_server_thread(upstream, serve)
         self.start_rtlmux(upstream_port, listen_port)
         self.assertTrue(ready.wait(2))
 
@@ -190,9 +194,7 @@ def serve():
             upstream.close()
             closed.set()
 
-        thread = threading.Thread(target=serve)
-        thread.start()
-        self.addCleanup(thread.join, 2)
+        self.start_server_thread(upstream, serve)
         process = self.start_rtlmux(upstream_port, listen_port)
 
         self.assertTrue(closed.wait(2))
@@ -222,9 +224,7 @@ def serve():
                 disconnected[index].set()
             upstream.close()
 
-        thread = threading.Thread(target=serve)
-        thread.start()
-        self.addCleanup(thread.join, 4)
+        self.start_server_thread(upstream, serve, 6)
         self.start_rtlmux(upstream_port, listen_port, "-d", "-r")
 
         first = connect_with_retry(listen_port)