Hide more interfaces behind the flow code.

adsbus/opts.c

@@ -97,7 +97,8 @@ bool opts_add_file_read(char *arg) {
 	if (fd == -1) {
 		return false;
 	}
-	receive_new(fd, NULL, NULL);
+	// TODO: add file.[ch]
+	receive_flow->new(fd, NULL, NULL);
 	return true;
 }
 
@@ -145,7 +146,8 @@ bool opts_add_exec_send(char *arg) {
 bool opts_add_stdin(char __attribute__((unused)) *arg) {
 	int fd = dup(0);
 	assert(!fcntl(fd, F_SETFD, FD_CLOEXEC));
-	receive_new(fd, NULL, NULL);
+	// TODO: add file.[ch]
+	receive_flow->new(fd, NULL, NULL);
 	return true;
 }
 

adsbus/receive.c

@@ -37,6 +37,8 @@ struct receive {
 };
 static struct receive *receive_head = NULL;
 
+static void receive_new(int, void *, struct peer *);
+
 static struct flow _receive_flow = {
 	.name = "receive",
 	.new = receive_new,
@@ -132,13 +134,7 @@ static void receive_read(struct peer *peer) {
 	}
 }
 
-void receive_cleanup() {
+static void receive_new(int fd, void __attribute__((unused)) *passthrough, struct peer *on_close) {
-	while (receive_head) {
-		receive_del(receive_head);
-	}
-}
-
-void receive_new(int fd, void __attribute__((unused)) *passthrough, struct peer *on_close) {
 	peer_count_in++;
 
 	socket_receive_init(fd);
@@ -164,6 +160,12 @@ void receive_new(int fd, void __attribute__((unused)) *passthrough, struct peer
 	fprintf(stderr, "R %s: New receive connection\n", receive->id);
 }
 
+void receive_cleanup() {
+	while (receive_head) {
+		receive_del(receive_head);
+	}
+}
+
 void receive_print_usage() {
 	fprintf(stderr, "\nSupported receive formats (auto-detected):\n");
 	for (size_t i = 0; i < NUM_PARSERS; i++) {

adsbus/receive.h

@@ -3,9 +3,7 @@
 #define PARSER_STATE_LEN 256
 
 struct flow;
-struct peer;
 
 void receive_cleanup(void);
-void receive_new(int, void *, struct peer *);
 void receive_print_usage(void);
 extern struct flow *receive_flow;

adsbus/send.c

@@ -33,10 +33,13 @@ struct send {
 	struct list_head send_list;
 };
 
+static void send_new(int, void *, struct peer *);
+static void send_get_hello(struct buf **, void *);
+
 static struct flow _send_flow = {
 	.name = "send",
-	.new = send_new_wrapper,
+	.new = send_new,
-	.get_hello = send_hello,
+	.get_hello = send_get_hello,
 	.ref_count = &peer_count_out,
 };
 struct flow *send_flow = &_send_flow;
@@ -96,32 +99,9 @@ static void send_del_wrapper(struct peer *peer) {
 	send_del((struct send *) peer);
 }
 
-void send_init() {
+static void send_new(int fd, void *passthrough, struct peer *on_close) {
-	assert(signal(SIGPIPE, SIG_IGN) != SIG_ERR);
+	struct serializer *serializer = (struct serializer *) passthrough;
-	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
-		list_head_init(&serializers[i].send_head);
-	}
-}
 
-void send_cleanup() {
-	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
-		struct send *iter, *next;
-		list_for_each_entry_safe(iter, next, &serializers[i].send_head, send_list) {
-			send_del(iter);
-		}
-	}
-}
-
-struct serializer *send_get_serializer(char *name) {
-	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
-		if (strcasecmp(serializers[i].name, name) == 0) {
-			return &serializers[i];
-		}
-	}
-	return NULL;
-}
-
-void send_new(int fd, struct serializer *serializer, struct peer *on_close) {
 	peer_count_out++;
 
 	socket_send_init(fd);
@@ -141,30 +121,51 @@ void send_new(int fd, struct serializer *serializer, struct peer *on_close) {
 	fprintf(stderr, "S %s (%s): New send connection\n", send->id, serializer->name);
 }
 
-void send_new_wrapper(int fd, void *passthrough, struct peer *on_close) {
+static void send_get_hello(struct buf **buf_pp, void *passthrough) {
-	send_new(fd, (struct serializer *) passthrough, on_close);
+	struct serializer *serializer = (struct serializer *) passthrough;
+	if (serializer->hello) {
+		serializer->hello(buf_pp);
+	}
 }
 
-bool send_new_hello(int fd, struct serializer *serializer, struct peer *on_close) {
+void send_init() {
+	assert(signal(SIGPIPE, SIG_IGN) != SIG_ERR);
+	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
+		list_head_init(&serializers[i].send_head);
+	}
+}
+
+void send_cleanup() {
+	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
+		struct send *iter, *next;
+		list_for_each_entry_safe(iter, next, &serializers[i].send_head, send_list) {
+			send_del(iter);
+		}
+	}
+}
+
+void *send_get_serializer(char *name) {
+	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
+		if (strcasecmp(serializers[i].name, name) == 0) {
+			return &serializers[i];
+		}
+	}
+	return NULL;
+}
+
+bool send_new_hello(int fd, void *passthrough, struct peer *on_close) {
 	struct buf buf = BUF_INIT, *buf_ptr = &buf;
-	send_hello(&buf_ptr, serializer);
+	send_get_hello(&buf_ptr, passthrough);
 	if (buf_ptr->length) {
 		if (write(fd, buf_at(buf_ptr, 0), buf_ptr->length) != (ssize_t) buf_ptr->length) {
 			assert(!close(fd));
 			return false;
 		}
 	}
-	send_new(fd, serializer, on_close);
+	send_new(fd, passthrough, on_close);
 	return true;
 }
 
-void send_hello(struct buf **buf_pp, void *passthrough) {
-	struct serializer *serializer = (struct serializer *) passthrough;
-	if (serializer->hello) {
-		serializer->hello(buf_pp);
-	}
-}
-
 void send_write(struct packet *packet) {
 	packet_sanity_check(packet);
 	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {

adsbus/send.h

@@ -2,18 +2,14 @@
 
 #include <stdbool.h>
 
-struct buf;
 struct flow;
 struct packet;
 struct peer;
 
 void send_init(void);
 void send_cleanup(void);
-struct serializer *send_get_serializer(char *);
+void *send_get_serializer(char *);
-void send_new(int, struct serializer *, struct peer *);
+bool send_new_hello(int, void *, struct peer *);
-void send_new_wrapper(int, void *, struct peer *);
-bool send_new_hello(int, struct serializer *, struct peer *);
-void send_hello(struct buf **, void *);
 void send_write(struct packet *);
 void send_print_usage(void);
 extern struct flow *send_flow;