diff --git a/adsbus/Makefile b/adsbus/Makefile
index 0297916..beec994 100644
--- a/adsbus/Makefile
+++ b/adsbus/Makefile
@@ -10,7 +10,8 @@ VALGRIND ?= valgrind
 VALGRIND_FLAGS ?= --error-exitcode=1 --trace-children=yes --track-fds=yes --show-leak-kinds=all --leak-check=full
 ADSBUS_TEST_FLAGS ?= --stdin --stdout=airspy_adsb --stdout=beast --stdout=json --stdout=proto --stdout=raw --stdout=stats
 
-OBJ_TRANSPORT = exec.o incoming.o outgoing.o receive.o send.o
+OBJ_TRANSPORT = exec.o file.o incoming.o outgoing.o
+OBJ_FLOW = flow.o receive.o send.o
 OBJ_PROTOCOL = airspy_adsb.o beast.o json.o proto.o raw.o stats.o
 OBJ_UTIL = buf.o hex.o list.o opts.o packet.o peer.o rand.o resolve.o server.o socket.o uuid.o wakeup.o
 OBJ_PROTO = adsb.pb-c.o
@@ -26,8 +27,8 @@ clean:
 adsb.pb-c.c: ../proto/adsb.proto
 	protoc-c --c_out=./ --proto_path=$(dir $<) $<
 
-adsbus: adsbus.o $(OBJ_TRANSPORT) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO)
-	$(COMP) $(LDFLAGS) -o adsbus adsbus.o $(OBJ_TRANSPORT) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO) $(LIBS)
+adsbus: adsbus.o $(OBJ_TRANSPORT) $(OBJ_FLOW) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO)
+	$(COMP) $(LDFLAGS) -o adsbus adsbus.o $(OBJ_TRANSPORT) $(OBJ_FLOW) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO) $(LIBS)
 
 fuzz:
 	rm -rf findings
diff --git a/adsbus/exec.c b/adsbus/exec.c
index 93770aa..426fa0e 100644
--- a/adsbus/exec.c
+++ b/adsbus/exec.c
@@ -60,23 +60,11 @@ static void exec_close_handler(struct peer *peer) {
 	wakeup_add((struct peer *) exec, delay);
 }
 
-static bool exec_hello(int fd, struct exec *exec) {
-	if (!exec->flow->get_hello) {
-		return true;
-	}
-	struct buf buf = BUF_INIT, *buf_ptr = &buf;
-	exec->flow->get_hello(&buf_ptr, exec->passthrough);
-	if (!buf_ptr->length) {
-		return true;
-	}
-	return (write(fd, buf_at(buf_ptr, 0), buf_ptr->length) == (ssize_t) buf_ptr->length);
-}
-
 static void exec_parent(struct exec *exec, pid_t child, int fd) {
 	exec->child = child;
 	fprintf(stderr, "E %s: Child started as process %d\n", exec->id, exec->child);
 
-	if (!exec_hello(fd, exec)) {
+	if (!flow_hello(fd, exec->flow, exec->passthrough)) {
 		assert(!close(fd));
 		exec_close_handler((struct peer *) exec);
 		return;
diff --git a/adsbus/file.c b/adsbus/file.c
new file mode 100644
index 0000000..7844fac
--- /dev/null
+++ b/adsbus/file.c
@@ -0,0 +1,43 @@
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include "flow.h"
+#include "receive.h"
+#include "send.h"
+
+#include "file.h"
+
+void file_fd_new(int fd, struct flow *flow, void *passthrough) {
+	flow->new(fd, passthrough, NULL);
+	// TODO: log error; retry?
+	flow_hello(fd, flow, passthrough);
+}
+
+void file_read_new(char *path, struct flow *flow, void *passthrough) {
+	int fd = open(path, O_RDONLY | O_CLOEXEC);
+	if (fd == -1) {
+		// TODO: log error; retry?
+		return;
+	}
+	file_fd_new(fd, flow, passthrough);
+}
+
+void file_write_new(char *path, struct flow *flow, void *passthrough) {
+	int fd = open(path, O_WRONLY | O_CREAT | O_NOFOLLOW | O_TRUNC | O_CLOEXEC, S_IRUSR | S_IWUSR);
+	if (fd == -1) {
+		// TODO: log error; retry?
+		return;
+	}
+	file_fd_new(fd, flow, passthrough);
+}
+
+void file_append_new(char *path, struct flow *flow, void *passthrough) {
+	int fd = open(path, O_WRONLY | O_CREAT | O_NOFOLLOW | O_CLOEXEC, S_IRWXU);
+	if (fd == -1) {
+		// TODO: log error; retry?
+		return;
+	}
+	flow->new(fd, flow, passthrough);
+}
diff --git a/adsbus/file.h b/adsbus/file.h
new file mode 100644
index 0000000..7053e91
--- /dev/null
+++ b/adsbus/file.h
@@ -0,0 +1,8 @@
+#pragma once
+
+struct flow;
+
+void file_fd_new(int, struct flow *, void *);
+void file_read_new(char *, struct flow *, void *);
+void file_write_new(char *, struct flow *, void *);
+void file_append_new(char *, struct flow *, void *);
diff --git a/adsbus/flow.c b/adsbus/flow.c
new file mode 100644
index 0000000..4aa46a7
--- /dev/null
+++ b/adsbus/flow.c
@@ -0,0 +1,17 @@
+#include <unistd.h>
+
+#include "buf.h"
+
+#include "flow.h"
+
+bool flow_hello(int fd, struct flow *flow, void *passthrough) {
+	if (!flow->get_hello) {
+		return true;
+	}
+	struct buf buf = BUF_INIT, *buf_ptr = &buf;
+	flow->get_hello(&buf_ptr, passthrough);
+	if (!buf_ptr->length) {
+		return true;
+	}
+	return (write(fd, buf_at(buf_ptr, 0), buf_ptr->length) == (ssize_t) buf_ptr->length);
+}
diff --git a/adsbus/flow.h b/adsbus/flow.h
index e7ff369..dfe9d07 100644
--- a/adsbus/flow.h
+++ b/adsbus/flow.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <stdbool.h>
 #include <stdint.h>
 
 struct buf;
@@ -15,3 +16,5 @@ struct flow {
 	flow_get_hello get_hello;
 	uint32_t *ref_count;
 };
+
+bool flow_hello(int, struct flow *, void *);
diff --git a/adsbus/incoming.c b/adsbus/incoming.c
index 74603b9..62da1bd 100644
--- a/adsbus/incoming.c
+++ b/adsbus/incoming.c
@@ -43,18 +43,6 @@ static void incoming_retry(struct incoming *incoming) {
 	wakeup_add((struct peer *) incoming, delay);
 }
 
-static bool incoming_hello(int fd, struct incoming *incoming) {
-	if (!incoming->flow->get_hello) {
-		return true;
-	}
-	struct buf buf = BUF_INIT, *buf_ptr = &buf;
-	incoming->flow->get_hello(&buf_ptr, incoming->passthrough);
-	if (!buf_ptr->length) {
-		return true;
-	}
-	return (write(fd, buf_at(buf_ptr, 0), buf_ptr->length) == (ssize_t) buf_ptr->length);
-}
-
 static void incoming_handler(struct peer *peer) {
 	struct incoming *incoming = (struct incoming *) peer;
 
@@ -78,7 +66,7 @@ static void incoming_handler(struct peer *peer) {
 			local_hbuf, local_sbuf,
 			peer_hbuf, peer_sbuf);
 
-	if (!incoming_hello(fd, incoming)) {
+	if (!flow_hello(fd, incoming->flow, incoming->passthrough)) {
 		fprintf(stderr, "I %s: Error writing greeting\n", incoming->id);
 		assert(!close(fd));
 		return;
diff --git a/adsbus/opts.c b/adsbus/opts.c
index 694af5e..9d5ab71 100644
--- a/adsbus/opts.c
+++ b/adsbus/opts.c
@@ -7,7 +7,7 @@
 #include <unistd.h>
 
 #include "exec.h"
-#include "flow.h"
+#include "file.h"
 #include "incoming.h"
 #include "outgoing.h"
 #include "receive.h"
@@ -93,39 +93,28 @@ bool opts_add_listen_send(char *arg) {
 }
 
 bool opts_add_file_read(char *arg) {
-	int fd = open(arg, O_RDONLY | O_CLOEXEC);
-	if (fd == -1) {
-		return false;
-	}
-	// TODO: add file.[ch]
-	receive_flow->new(fd, NULL, NULL);
+	file_read_new(arg, receive_flow, NULL);
 	return true;
 }
 
 bool opts_add_file_write(char *arg) {
 	struct serializer *serializer = opts_get_serializer(&arg);
 	if (!serializer) {
-		return NULL;
-	}
-
-	int fd = open(arg, O_WRONLY | O_CREAT | O_NOFOLLOW | O_TRUNC | O_CLOEXEC, S_IRWXU);
-	if (fd == -1) {
 		return false;
 	}
-	return send_new_hello(fd, serializer, NULL);
+
+	file_write_new(arg, send_flow, serializer);
+	return true;
 }
 
 bool opts_add_file_append(char *arg) {
 	struct serializer *serializer = opts_get_serializer(&arg);
 	if (!serializer) {
-		return NULL;
-	}
-
-	int fd = open(arg, O_WRONLY | O_CREAT | O_NOFOLLOW | O_CLOEXEC, S_IRWXU);
-	if (fd == -1) {
 		return false;
 	}
-	return send_new_hello(fd, serializer, NULL);
+
+	file_append_new(arg, send_flow, serializer);
+	return true;
 }
 
 bool opts_add_exec_receive(char *arg) {
@@ -136,7 +125,7 @@ bool opts_add_exec_receive(char *arg) {
 bool opts_add_exec_send(char *arg) {
 	struct serializer *serializer = opts_get_serializer(&arg);
 	if (!serializer) {
-		return NULL;
+		return false;
 	}
 
 	exec_new(arg, send_flow, serializer);
@@ -146,8 +135,7 @@ 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));
-	// TODO: add file.[ch]
-	receive_flow->new(fd, NULL, NULL);
+	file_fd_new(fd, receive_flow, NULL);
 	return true;
 }
 
@@ -158,5 +146,6 @@ bool opts_add_stdout(char *arg) {
 	}
 	int fd = dup(1);
 	assert(!fcntl(fd, F_SETFD, FD_CLOEXEC));
-	return send_new_hello(fd, serializer, NULL);
+	file_fd_new(fd, send_flow, serializer);
+	return true;
 }
diff --git a/adsbus/send.c b/adsbus/send.c
index cd9a56d..6d3cd7f 100644
--- a/adsbus/send.c
+++ b/adsbus/send.c
@@ -153,19 +153,6 @@ void *send_get_serializer(char *name) {
 	return NULL;
 }
 
-bool send_new_hello(int fd, void *passthrough, struct peer *on_close) {
-	struct buf buf = BUF_INIT, *buf_ptr = &buf;
-	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, passthrough, on_close);
-	return true;
-}
-
 void send_write(struct packet *packet) {
 	packet_sanity_check(packet);
 	for (size_t i = 0; i < NUM_SERIALIZERS; i++) {
diff --git a/adsbus/send.h b/adsbus/send.h
index 17316cc..eab1c4e 100644
--- a/adsbus/send.h
+++ b/adsbus/send.h
@@ -9,7 +9,6 @@ struct peer;
 void send_init(void);
 void send_cleanup(void);
 void *send_get_serializer(char *);
-bool send_new_hello(int, void *, struct peer *);
 void send_write(struct packet *);
 void send_print_usage(void);
 extern struct flow *send_flow;