Factor out opts parsing, clean up shutdown.

adsbus/Makefile

@@ -11,5 +11,5 @@ clean:
 %.o: %.c *.h
 	$(CC) -c $(CFLAGS) $< -o $@
 
-adsbus: adsbus.o receive.o send.o incoming.o outgoing.o beast.o json.o raw.o stats.o wakeup.o common.o
+adsbus: adsbus.o receive.o send.o incoming.o outgoing.o beast.o json.o raw.o stats.o wakeup.o opts.o common.o
-	$(CC) $(LDFLAGS) -o adsbus adsbus.o receive.o send.o incoming.o outgoing.o beast.o json.o raw.o stats.o wakeup.o common.o $(LIBS)
+	$(CC) $(LDFLAGS) -o adsbus adsbus.o receive.o send.o incoming.o outgoing.o beast.o json.o raw.o stats.o wakeup.o opts.o common.o $(LIBS)

adsbus/adsbus.c

@@ -9,13 +9,12 @@
 #include "receive.h"
 #include "send.h"
 
-#include "incoming.h"
-#include "outgoing.h"
-
 #include "beast.h"
 #include "json.h"
 #include "stats.h"
 
+#include "opts.h"
+
 static void print_usage(const char *name) {
 	fprintf(stderr,
 			"\n"
@@ -33,88 +32,6 @@ static void print_usage(const char *name) {
 	send_print_usage();
 }
 
-static bool add_dump(char *arg) {
-	struct serializer *serializer = send_get_serializer(arg);
-	if (!serializer) {
-		fprintf(stderr, "Unknown --dump=FORMAT: %s\n", arg);
-		return false;
-	}
-	send_add(1, serializer);
-	return true;
-}
-
-static bool add_connect_receive(char *arg) {
-	char *port = strrchr(arg, '/');
-	if (!port) {
-		fprintf(stderr, "Invalid --connect-receive=HOST/PORT (missing \"/\"): %s\n", arg);
-		return false;
-	}
-	*(port++) = '\0';
-
-	outgoing_new(arg, port, receive_new, NULL);
-	return true;
-}
-
-static bool add_connect_send(char *arg) {
-	char *host_port = strchr(arg, '=');
-	if (!host_port) {
-		fprintf(stderr, "Invalid --connect-send=FORMAT=HOST/PORT (missing \"=\"): %s\n", arg);
-		return false;
-	}
-	*(host_port++) = '\0';
-
-	struct serializer *serializer = send_get_serializer(arg);
-	if (!serializer) {
-		fprintf(stderr, "Unknown --connect-send=FORMAT=HOST/PORT format: %s\n", arg);
-		return false;
-	}
-
-	char *port = strrchr(host_port, '/');
-	if (!port) {
-		fprintf(stderr, "Invalid --connect-send=FORMAT=HOST/PORT (missing \"/\"): %s\n", host_port);
-		return false;
-	}
-	*(port++) = '\0';
-
-	incoming_new(host_port, port, send_add_wrapper, serializer);
-	return true;
-}
-
-static bool add_listen_receive(char *arg){
-	char *port = strrchr(arg, '/');
-	if (port) {
-		*(port++) = '\0';
-		incoming_new(arg, port, receive_new, NULL);
-	} else {
-		incoming_new(NULL, arg, receive_new, NULL);
-	}
-	return true;
-}
-
-static bool add_listen_send(char *arg) {
-	char *host_port = strchr(arg, '=');
-	if (!host_port) {
-		fprintf(stderr, "Invalid --listen-send=FORMAT=[HOST/]PORT (missing \"=\"): %s\n", arg);
-		return false;
-	}
-	*(host_port++) = '\0';
-
-	struct serializer *serializer = send_get_serializer(arg);
-	if (!serializer) {
-		fprintf(stderr, "Unknown --listen-send=FORMAT=[HOST/]PORT format: %s\n", arg);
-		return false;
-	}
-
-	char *port = strrchr(host_port, '/');
-	if (port) {
-		*(port++) = '\0';
-		incoming_new(host_port, port, send_add_wrapper, serializer);
-	} else {
-		incoming_new(NULL, host_port, send_add_wrapper, serializer);
-	}
-	return true;
-}
-
 static bool parse_opts(int argc, char *argv[]) {
 	static struct option long_options[] = {
 		{"dump",            required_argument, 0, 'd'},
@@ -131,23 +48,23 @@ static bool parse_opts(int argc, char *argv[]) {
 		bool (*handler)(char *) = NULL;
 		switch (opt) {
 		  case 'd':
-				handler = add_dump;
+				handler = opts_add_dump;
 				break;
 
 			case 'c':
-				handler = add_connect_receive;
+				handler = opts_add_connect_receive;
 				break;
 
 			case 's':
-				handler = add_connect_send;
+				handler = opts_add_connect_send;
 				break;
 
 			case 'l':
-				handler = add_listen_receive;
+				handler = opts_add_listen_receive;
 				break;
 
 			case 'm':
-				handler = add_listen_send;
+				handler = opts_add_listen_send;
 				break;
 
 			case 'h':
@@ -158,6 +75,7 @@ static bool parse_opts(int argc, char *argv[]) {
 
 		if (handler) {
 			if (!handler(optarg)) {
+				fprintf(stderr, "Invalid flag value: %s\n", optarg);
 				print_usage(argv[0]);
 				return false;
 			}
@@ -190,5 +108,7 @@ int main(int argc, char *argv[]) {
 	}
 
 	peer_loop();
+
+	send_cleanup();
 	return EXIT_SUCCESS;
 }

adsbus/common.c

@@ -1,16 +1,24 @@
 #include <stdio.h>
+#include <stdbool.h>
 #include <assert.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
+#include <signal.h>
 #include <uuid/uuid.h>
 
 #include "common.h"
 
 
-int epoll_fd;
+static int epoll_fd;
+static bool peer_canceled = false;
+
+static void peer_cancel(int signal) {
+	peer_canceled = true;
+}
 
 void peer_init() {
+	signal(SIGINT, peer_cancel);
 	epoll_fd = epoll_create1(0);
 	assert(epoll_fd >= 0);
 }
@@ -30,11 +38,14 @@ void peer_epoll_del(struct peer *peer) {
 }
 
 void peer_loop() {
-	while (1) {
+	while (!peer_canceled) {
 #define MAX_EVENTS 10
 		struct epoll_event events[MAX_EVENTS];
 		int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
-		assert(nfds >= 0);
+		if (nfds < 0) {
+			perror("epoll_wait");
+			break;
+		}
 
     for (int n = 0; n < nfds; n++) {
 			struct peer *peer = events[n].data.ptr;

adsbus/common.h

@@ -13,7 +13,6 @@ struct peer {
 	int fd;
 	peer_event_handler event_handler;
 };
-extern int epoll_fd;
 void peer_init();
 void peer_epoll_add(struct peer *, uint32_t);
 void peer_epoll_del(struct peer *);

adsbus/incoming.c

@@ -52,8 +52,8 @@ void incoming_new(const char *node, const char *service, incoming_connection_han
 	struct incoming *incoming = malloc(sizeof(*incoming));
 	incoming->peer.event_handler = incoming_handler;
 	uuid_gen(incoming->id);
-	incoming->node = node;
+	incoming->node = strdup(node);
-	incoming->service = service;
+	incoming->service = strdup(service);
 	incoming->handler = handler;
 	incoming->passthrough = passthrough;
 
@@ -69,6 +69,7 @@ void incoming_new(const char *node, const char *service, incoming_connection_han
 	int gai_err = getaddrinfo(incoming->node, incoming->service, &hints, &addrs);
 	if (gai_err) {
 		fprintf(stderr, "I %s: Failed to resolve %s/%s: %s\n", incoming->id, incoming->node, incoming->service, gai_strerror(gai_err));
+		// TODO: add incoming_del, free strdup values
 		free(incoming);
 		return;
 	}
@@ -99,6 +100,7 @@ void incoming_new(const char *node, const char *service, incoming_connection_han
 
 	if (addr == NULL) {
 		fprintf(stderr, "I %s: Failed to bind any addresses for %s/%s...\n", incoming->id, incoming->node, incoming->service);
+		// TODO: use incoming_del
 		free(incoming);
 		return;
 	}

adsbus/opts.c

@@ -0,0 +1,97 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "incoming.h"
+#include "outgoing.h"
+
+#include "receive.h"
+#include "send.h"
+
+#include "opts.h"
+
+static char *opts_split(char **arg, char delim) {
+	char *split = strchr(*arg, delim);
+	if (!split) {
+		return NULL;
+	}
+	char *ret = strndup(*arg, split - *arg);
+	*arg = split + 1;
+	return ret;
+}
+
+bool opts_add_dump(char *arg) {
+	struct serializer *serializer = send_get_serializer(arg);
+	if (!serializer) {
+		return false;
+	}
+	send_add(1, serializer);
+	return true;
+}
+
+bool opts_add_connect_receive(char *arg) {
+	char *host = opts_split(&arg, '/');
+	if (!host) {
+		return false;
+	}
+
+	outgoing_new(host, arg, receive_new, NULL);
+	free(host);
+	return true;
+}
+
+bool opts_add_connect_send(char *arg) {
+	char *format = opts_split(&arg, '=');
+	if (!format) {
+		return false;
+	}
+
+	struct serializer *serializer = send_get_serializer(format);
+	if (!serializer) {
+		free(format);
+		return false;
+	}
+
+	char *host = opts_split(&arg, '/');
+	if (!host) {
+		free(format);
+		return false;
+	}
+
+	incoming_new(host, arg, send_add_wrapper, serializer);
+	free(format);
+	free(host);
+	return true;
+}
+
+bool opts_add_listen_receive(char *arg) {
+	char *host = opts_split(&arg, '/');
+	if (host) {
+		incoming_new(host, arg, receive_new, NULL);
+		free(host);
+	} else {
+		incoming_new(NULL, arg, receive_new, NULL);
+	}
+	return true;
+}
+
+bool opts_add_listen_send(char *arg) {
+	char *format = opts_split(&arg, '=');
+	if (!format) {
+		return false;
+	}
+
+	struct serializer *serializer = send_get_serializer(format);
+	if (!serializer) {
+		free(format);
+		return false;
+	}
+
+	char *host = opts_split(&arg, '/');
+	if (host) {
+		incoming_new(host, arg, send_add_wrapper, serializer);
+		free(host);
+	} else {
+		incoming_new(NULL, arg, send_add_wrapper, serializer);
+	}
+	return true;
+}

adsbus/opts.h

@@ -0,0 +1,9 @@
+#pragma once
+
+#include <stdbool.h>
+
+bool opts_add_dump(char *);
+bool opts_add_connect_receive(char *);
+bool opts_add_connect_send(char *);
+bool opts_add_listen_receive(char *);
+bool opts_add_listen_send(char *);

adsbus/outgoing.c

@@ -113,8 +113,8 @@ static void outgoing_resolve(struct outgoing *outgoing) {
 void outgoing_new(const char *node, const char *service, outgoing_connection_handler handler, void *passthrough) {
 	struct outgoing *outgoing = malloc(sizeof(*outgoing));
 	uuid_gen(outgoing->id);
-	outgoing->node = node;
+	outgoing->node = strdup(node);
-	outgoing->service = service;
+	outgoing->service = strdup(service);
 	outgoing->handler = handler;
 	outgoing->passthrough = passthrough;
 	outgoing_resolve(outgoing);

adsbus/send.c

@@ -82,6 +82,18 @@ void send_init() {
 	signal(SIGPIPE, SIG_IGN);
 }
 
+void send_cleanup() {
+	for (int i = 0; i < NUM_SERIALIZERS; i++) {
+		struct serializer *serializer = &serializers[i];
+		struct send *send = serializer->send_head;
+		while (send) {
+			struct send *next = send->next;
+			free(send);
+			send = next;
+		}
+	}
+}
+
 struct serializer *send_get_serializer(char *name) {
 	for (int i = 0; i < NUM_SERIALIZERS; i++) {
 		if (strcasecmp(serializers[i].name, name) == 0) {

adsbus/send.h

@@ -3,6 +3,7 @@
 #include "common.h"
 
 void send_init();
+void send_cleanup();
 struct serializer *send_get_serializer(char *);
 void send_add(int, struct serializer *);
 void send_add_wrapper(int, void *);