Move more socket options into the flow framework.

adsbus/flow.c

@@ -12,6 +12,12 @@ void flow_socket_ready(int fd, struct flow *flow) {
 	}
 }
 
+void flow_socket_connected(int fd, struct flow *flow) {
+	if (flow->socket_connected) {
+		flow->socket_connected(fd);
+	}
+}
+
 bool flow_hello(int fd, struct flow *flow, void *passthrough) {
 	if (!flow->get_hello) {
 		return true;

adsbus/flow.h

@@ -9,12 +9,14 @@ struct peer;
 struct flow {
 	const char *name;
 	void (*socket_ready)(int);
+	void (*socket_connected)(int);
 	void (*new)(int, void *, struct peer *);
 	void (*get_hello)(struct buf **, void *);
 	uint32_t *ref_count;
 };
 
 void flow_socket_ready(int, struct flow *);
+void flow_socket_connected(int, struct flow *);
 bool flow_hello(int, struct flow *, void *);
 bool flow_new(int, struct flow *, void *);
 void flow_ref_inc(struct flow *);

adsbus/incoming.c

@@ -64,6 +64,8 @@ static void incoming_handler(struct peer *peer) {
 			local_hbuf, local_sbuf,
 			peer_hbuf, peer_sbuf);
 
+	flow_socket_connected(fd, incoming->flow);
+
 	if (!flow_hello(fd, incoming->flow, incoming->passthrough)) {
 		fprintf(stderr, "I %s: Error writing greeting\n", incoming->id);
 		assert(!close(fd));

adsbus/outgoing.c

@@ -95,11 +95,12 @@ static void outgoing_connect_result(struct outgoing *outgoing, int result) {
 		case 0:
 			fprintf(stderr, "O %s: Connected to %s/%s\n", outgoing->id, hbuf, sbuf);
 			freeaddrinfo(outgoing->addrs);
-			flow_socket_ready(outgoing->peer.fd, outgoing->flow);
 			outgoing->attempt = 0;
 			int fd = outgoing->peer.fd;
 			outgoing->peer.fd = -1;
 			outgoing->peer.event_handler = outgoing_disconnect_handler;
+			flow_socket_ready(fd, outgoing->flow);
+			flow_socket_connected(fd, outgoing->flow);
 			outgoing->flow->new(fd, outgoing->passthrough, (struct peer *) outgoing);
 			break;
 

adsbus/receive.c

@@ -41,6 +41,7 @@ static void receive_new(int, void *, struct peer *);
 
 static struct flow _receive_flow = {
 	.name = "receive",
+	.socket_connected = socket_connected_receive,
 	.new = receive_new,
 	.ref_count = &peer_count_in,
 };
@@ -133,8 +134,6 @@ static void receive_read(struct peer *peer) {
 static void receive_new(int fd, void __attribute__((unused)) *passthrough, struct peer *on_close) {
 	peer_count_in++;
 
-	socket_receive(fd);
-
 	struct receive *receive = malloc(sizeof(*receive));
 	assert(receive);
 

adsbus/send.c

@@ -39,6 +39,7 @@ static void send_get_hello(struct buf **, void *);
 static struct flow _send_flow = {
 	.name = "send",
 	.socket_ready = socket_ready_send,
+	.socket_connected = socket_connected_send,
 	.new = send_new,
 	.get_hello = send_get_hello,
 	.ref_count = &peer_count_out,
@@ -105,8 +106,6 @@ static void send_new(int fd, void *passthrough, struct peer *on_close) {
 
 	peer_count_out++;
 
-	socket_send(fd);
-
 	struct send *send = malloc(sizeof(*send));
 	assert(send);
 

adsbus/socket.c

@@ -48,13 +48,13 @@ void socket_ready_send(int fd) {
 	assert(res == 0);
 }
 
-void socket_send(int fd) {
+void socket_connected_send(int fd) {
 	// Called by data flow code; NOT safe to assume that fd is a socket
 	int res = shutdown(fd, SHUT_RD);
 	assert(res == 0 || (res == -1 && errno == ENOTSOCK));
 }
 
-void socket_receive(int fd) {
+void socket_connected_receive(int fd) {
 	// Called by data flow code; NOT safe to assume that fd is a socket
 	int res = shutdown(fd, SHUT_WR);
 	assert(res == 0 || (res == -1 && errno == ENOTSOCK));

adsbus/socket.h

@@ -4,5 +4,5 @@ void socket_pre_bind(int);
 void socket_pre_listen(int);
 void socket_ready(int);
 void socket_ready_send(int);
-void socket_send(int);
-void socket_receive(int);
+void socket_connected_send(int);
+void socket_connected_receive(int);