misc/adsb-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Encapsulate flow descriptions inside a struct.

Browse Source
This commit is contained in:
Ian Gulliver
2016-02-29 17:12:06 -08:00
parent 3539f9d640
commit 3fb3c02cf5
12 changed files with 83 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
adsbus/exec.c
View File

@@ -9,6 +9,7 @@
#include <unistd.h> #include <unistd.h>
#include "buf.h" #include "buf.h"
#include "flow.h"
#include "list.h" #include "list.h"
#include "peer.h" #include "peer.h"
#include "uuid.h" #include "uuid.h"
@@ -20,10 +21,8 @@ struct exec {
struct peer peer; struct peer peer;
uint8_t id[UUID_LEN]; uint8_t id[UUID_LEN];
char *command; char *command;
exec_connection_handler handler; struct flow *flow;
exec_get_hello hello;
void *passthrough; void *passthrough;
uint32_t *count;
pid_t child; pid_t child;
struct list_head exec_list; struct list_head exec_list;
}; };
@@ -33,7 +32,7 @@ static struct list_head exec_head = LIST_HEAD_INIT(exec_head);
static void exec_spawn_wrapper(struct peer *); static void exec_spawn_wrapper(struct peer *);
static void exec_del(struct exec *exec) { static void exec_del(struct exec *exec) {
(*exec->count)--; (*exec->flow->ref_count)--;
if (exec->child > 0) { if (exec->child > 0) {
fprintf(stderr, "E %s: Sending SIGTERM to child process %d\n", exec->id, exec->child); fprintf(stderr, "E %s: Sending SIGTERM to child process %d\n", exec->id, exec->child);
// Racy with the process terminating, so don't assert on it // Racy with the process terminating, so don't assert on it
@@ -62,11 +61,11 @@ static void exec_close_handler(struct peer *peer) {
} }
static bool exec_hello(int fd, struct exec *exec) { static bool exec_hello(int fd, struct exec *exec) {
if (!exec->hello) { if (!exec->flow->get_hello) {
return true; return true;
} }
struct buf buf = BUF_INIT, *buf_ptr = &buf; struct buf buf = BUF_INIT, *buf_ptr = &buf;
exec->hello(&buf_ptr, exec->passthrough); exec->flow->get_hello(&buf_ptr, exec->passthrough);
if (!buf_ptr->length) { if (!buf_ptr->length) {
return true; return true;
} }
@@ -84,7 +83,7 @@ static void exec_parent(struct exec *exec, pid_t child, int fd) {
} }
exec->peer.event_handler = exec_close_handler; exec->peer.event_handler = exec_close_handler;
exec->handler(fd, exec->passthrough, (struct peer *) exec); exec->flow->new(fd, exec->passthrough, (struct peer *) exec);
} }
static void __attribute__ ((noreturn)) exec_child(const struct exec *exec, int fd) { static void __attribute__ ((noreturn)) exec_child(const struct exec *exec, int fd) {
@@ -132,17 +131,15 @@ void exec_cleanup() {
} }
} }
void exec_new(char *command, exec_connection_handler handler, exec_get_hello hello, void *passthrough, uint32_t *count) { void exec_new(char *command, struct flow *flow, void *passthrough) {
(*count)++; (*flow->ref_count)++;
struct exec *exec = malloc(sizeof(*exec)); struct exec *exec = malloc(sizeof(*exec));
exec->peer.fd = -1; exec->peer.fd = -1;
uuid_gen(exec->id); uuid_gen(exec->id);
exec->command = strdup(command); exec->command = strdup(command);
exec->handler = handler; exec->flow = flow;
exec->hello = hello;
exec->passthrough = passthrough; exec->passthrough = passthrough;
exec->count = count;
list_add(&exec->exec_list, &exec_head); list_add(&exec->exec_list, &exec_head);

9
adsbus/exec.h
View File

@@ -1,11 +1,6 @@
#pragma once #pragma once
#include <stdint.h> struct flow;
struct buf;
struct peer;
void exec_cleanup(void); void exec_cleanup(void);
typedef void (*exec_connection_handler)(int fd, void *, struct peer *); void exec_new(char *, struct flow *, void *);
typedef void (*exec_get_hello)(struct buf **, void *);
void exec_new(char *, exec_connection_handler, exec_get_hello, void *, uint32_t *);

17
adsbus/flow.h Normal file
View File

@@ -0,0 +1,17 @@
#pragma once
#include <stdint.h>
struct buf;
struct peer;
typedef void (*flow_bound_socket_init)(int);
typedef void (*flow_new)(int, void *, struct peer *);
typedef void (*flow_get_hello)(struct buf **, void *);
struct flow {
const char *name;
flow_bound_socket_init bound_socket_init;
flow_new new;
flow_get_hello get_hello;
uint32_t *ref_count;
};

21
adsbus/incoming.c
View File

@@ -9,6 +9,7 @@
#include <unistd.h> #include <unistd.h>
#include "buf.h" #include "buf.h"
#include "flow.h"
#include "list.h" #include "list.h"
#include "peer.h" #include "peer.h"
#include "resolve.h" #include "resolve.h"
@@ -26,10 +27,8 @@ struct incoming {
struct addrinfo *addrs; struct addrinfo *addrs;
const char *error; const char *error;
uint32_t attempt; uint32_t attempt;
incoming_connection_handler handler; struct flow *flow;
incoming_get_hello hello;
void *passthrough; void *passthrough;
uint32_t *count;
struct list_head incoming_list; struct list_head incoming_list;
}; };
@@ -45,11 +44,11 @@ static void incoming_retry(struct incoming *incoming) {
} }
static bool incoming_hello(int fd, struct incoming *incoming) { static bool incoming_hello(int fd, struct incoming *incoming) {
if (!incoming->hello) { if (!incoming->flow->get_hello) {
return true; return true;
} }
struct buf buf = BUF_INIT, *buf_ptr = &buf; struct buf buf = BUF_INIT, *buf_ptr = &buf;
incoming->hello(&buf_ptr, incoming->passthrough); incoming->flow->get_hello(&buf_ptr, incoming->passthrough);
if (!buf_ptr->length) { if (!buf_ptr->length) {
return true; return true;
} }
@@ -85,11 +84,11 @@ static void incoming_handler(struct peer *peer) {
return; return;
} }
incoming->handler(fd, incoming->passthrough, NULL); incoming->flow->new(fd, incoming->passthrough, NULL);
} }
static void incoming_del(struct incoming *incoming) { static void incoming_del(struct incoming *incoming) {
(*incoming->count)--; (*incoming->flow->ref_count)--;
if (incoming->peer.fd >= 0) { if (incoming->peer.fd >= 0) {
assert(!close(incoming->peer.fd)); assert(!close(incoming->peer.fd));
} }
@@ -163,8 +162,8 @@ void incoming_cleanup() {
} }
} }
void incoming_new(char *node, char *service, incoming_connection_handler handler, incoming_get_hello hello, void *passthrough, uint32_t *count) { void incoming_new(char *node, char *service, struct flow *flow, void *passthrough) {
(*count)++; (*flow->ref_count)++;
struct incoming *incoming = malloc(sizeof(*incoming)); struct incoming *incoming = malloc(sizeof(*incoming));
incoming->peer.event_handler = incoming_handler; incoming->peer.event_handler = incoming_handler;
@@ -172,10 +171,8 @@ void incoming_new(char *node, char *service, incoming_connection_handler handler
incoming->node = node ? strdup(node) : NULL; incoming->node = node ? strdup(node) : NULL;
incoming->service = strdup(service); incoming->service = strdup(service);
incoming->attempt = 0; incoming->attempt = 0;
incoming->handler = handler; incoming->flow = flow;
incoming->hello = hello;
incoming->passthrough = passthrough; incoming->passthrough = passthrough;
incoming->count = count;
list_add(&incoming->incoming_list, &incoming_head); list_add(&incoming->incoming_list, &incoming_head);

9
adsbus/incoming.h
View File

@@ -1,11 +1,6 @@
#pragma once #pragma once
#include <stdint.h> struct flow;
struct buf;
struct peer;
void incoming_cleanup(void); void incoming_cleanup(void);
typedef void (*incoming_connection_handler)(int fd, void *, struct peer *); void incoming_new(char *, char *, struct flow *, void *);
typedef void (*incoming_get_hello)(struct buf **, void *);
void incoming_new(char *, char *, incoming_connection_handler, incoming_get_hello, void *, uint32_t *);

21
adsbus/opts.c
View File

@@ -6,11 +6,10 @@
#include <sys/types.h> #include <sys/types.h>
#include <unistd.h> #include <unistd.h>
#include "buf.h"
#include "exec.h" #include "exec.h"
#include "flow.h"
#include "incoming.h" #include "incoming.h"
#include "outgoing.h" #include "outgoing.h"
#include "peer.h"
#include "receive.h" #include "receive.h"
#include "send.h" #include "send.h"
@@ -26,13 +25,13 @@ static char *opts_split(char **arg, char delim) {
return ret; return ret;
} }
static void opts_add_listen(char *host_port, incoming_connection_handler handler, incoming_get_hello hello, void *passthrough, uint32_t *count) { static void opts_add_listen(char *host_port, struct flow *flow, void *passthrough) {
char *host = opts_split(&host_port, '/'); char *host = opts_split(&host_port, '/');
if (host) { if (host) {
incoming_new(host, host_port, handler, hello, passthrough, count); incoming_new(host, host_port, flow, passthrough);
free(host); free(host);
} else { } else {
incoming_new(NULL, host_port, handler, hello, passthrough, count); incoming_new(NULL, host_port, flow, passthrough);
} }
} }
@@ -57,7 +56,7 @@ bool opts_add_connect_receive(char *arg) {
return false; return false;
} }
outgoing_new(host, arg, receive_new, NULL, NULL, &peer_count_in); outgoing_new(host, arg, receive_flow, NULL);
free(host); free(host);
return true; return true;
} }
@@ -73,13 +72,13 @@ bool opts_add_connect_send(char *arg) {
return false; return false;
} }
outgoing_new(host, arg, send_new_wrapper, send_hello, serializer, &peer_count_out); outgoing_new(host, arg, send_flow, serializer);
free(host); free(host);
return true; return true;
} }
bool opts_add_listen_receive(char *arg) { bool opts_add_listen_receive(char *arg) {
opts_add_listen(arg, receive_new, NULL, NULL, &peer_count_in); opts_add_listen(arg, receive_flow, NULL);
return true; return true;
} }
@@ -89,7 +88,7 @@ bool opts_add_listen_send(char *arg) {
return false; return false;
} }
opts_add_listen(arg, send_new_wrapper, send_hello, serializer, &peer_count_out); opts_add_listen(arg, send_flow, serializer);
return true; return true;
} }
@@ -129,7 +128,7 @@ bool opts_add_file_append(char *arg) {
} }
bool opts_add_exec_receive(char *arg) { bool opts_add_exec_receive(char *arg) {
exec_new(arg, receive_new, NULL, NULL, &peer_count_in); exec_new(arg, receive_flow, NULL);
return true; return true;
} }
@@ -139,7 +138,7 @@ bool opts_add_exec_send(char *arg) {
return NULL; return NULL;
} }
exec_new(arg, send_new_wrapper, send_hello, serializer, &peer_count_out); exec_new(arg, send_flow, serializer);
return true; return true;
} }

21
adsbus/outgoing.c
View File

@@ -9,6 +9,7 @@
#include <unistd.h> #include <unistd.h>
#include "buf.h" #include "buf.h"
#include "flow.h"
#include "list.h" #include "list.h"
#include "peer.h" #include "peer.h"
#include "resolve.h" #include "resolve.h"
@@ -27,10 +28,8 @@ struct outgoing {
struct addrinfo *addr; struct addrinfo *addr;
const char *error; const char *error;
uint32_t attempt; uint32_t attempt;
outgoing_connection_handler handler; struct flow *flow;
outgoing_get_hello hello;
void *passthrough; void *passthrough;
uint32_t *count;
struct list_head outgoing_list; struct list_head outgoing_list;
}; };
@@ -63,8 +62,8 @@ static void outgoing_connect_next(struct outgoing *outgoing) {
assert(outgoing->peer.fd >= 0); assert(outgoing->peer.fd >= 0);
struct buf buf = BUF_INIT, *buf_ptr = &buf; struct buf buf = BUF_INIT, *buf_ptr = &buf;
if (outgoing->hello) { if (outgoing->flow->get_hello) {
outgoing->hello(&buf_ptr, outgoing->passthrough); outgoing->flow->get_hello(&buf_ptr, outgoing->passthrough);
} }
ssize_t result = sendto(outgoing->peer.fd, buf_at(buf_ptr, 0), buf_ptr->length, MSG_FASTOPEN, outgoing->addr->ai_addr, outgoing->addr->ai_addrlen); ssize_t result = sendto(outgoing->peer.fd, buf_at(buf_ptr, 0), buf_ptr->length, MSG_FASTOPEN, outgoing->addr->ai_addr, outgoing->addr->ai_addrlen);
outgoing_connect_result(outgoing, result == (ssize_t) buf_ptr->length ? EINPROGRESS : errno); outgoing_connect_result(outgoing, result == (ssize_t) buf_ptr->length ? EINPROGRESS : errno);
@@ -102,7 +101,7 @@ static void outgoing_connect_result(struct outgoing *outgoing, int result) {
int fd = outgoing->peer.fd; int fd = outgoing->peer.fd;
outgoing->peer.fd = -1; outgoing->peer.fd = -1;
outgoing->peer.event_handler = outgoing_disconnect_handler; outgoing->peer.event_handler = outgoing_disconnect_handler;
outgoing->handler(fd, outgoing->passthrough, (struct peer *) outgoing); outgoing->flow->new(fd, outgoing->passthrough, (struct peer *) outgoing);
break; break;
case EINPROGRESS: case EINPROGRESS:
@@ -144,7 +143,7 @@ static void outgoing_resolve_wrapper(struct peer *peer) {
} }
static void outgoing_del(struct outgoing *outgoing) { static void outgoing_del(struct outgoing *outgoing) {
(*outgoing->count)--; (*outgoing->flow->ref_count)--;
if (outgoing->peer.fd >= 0) { if (outgoing->peer.fd >= 0) {
assert(!close(outgoing->peer.fd)); assert(!close(outgoing->peer.fd));
} }
@@ -160,18 +159,16 @@ void outgoing_cleanup() {
} }
} }
void outgoing_new(char *node, char *service, outgoing_connection_handler handler, outgoing_get_hello hello, void *passthrough, uint32_t *count) { void outgoing_new(char *node, char *service, struct flow *flow, void *passthrough) {
(*count)++; (*flow->ref_count)++;
struct outgoing *outgoing = malloc(sizeof(*outgoing)); struct outgoing *outgoing = malloc(sizeof(*outgoing));
uuid_gen(outgoing->id); uuid_gen(outgoing->id);
outgoing->node = strdup(node); outgoing->node = strdup(node);
outgoing->service = strdup(service); outgoing->service = strdup(service);
outgoing->attempt = 0; outgoing->attempt = 0;
outgoing->handler = handler; outgoing->flow = flow;
outgoing->hello = hello;
outgoing->passthrough = passthrough; outgoing->passthrough = passthrough;
outgoing->count = count;
list_add(&outgoing->outgoing_list, &outgoing_head); list_add(&outgoing->outgoing_list, &outgoing_head);

7
adsbus/outgoing.h
View File

@@ -1,9 +1,6 @@
#pragma once #pragma once
struct buf; struct flow;
struct peer;
void outgoing_cleanup(void); void outgoing_cleanup(void);
typedef void (*outgoing_connection_handler)(int fd, void *, struct peer *); void outgoing_new(char *, char *, struct flow *, void *);
typedef void (*outgoing_get_hello)(struct buf **, void *);
void outgoing_new(char *, char *, outgoing_connection_handler, outgoing_get_hello, void *, uint32_t *);

8
adsbus/receive.c
View File

@@ -9,6 +9,7 @@
#include "airspy_adsb.h" #include "airspy_adsb.h"
#include "beast.h" #include "beast.h"
#include "buf.h" #include "buf.h"
#include "flow.h"
#include "json.h" #include "json.h"
#include "packet.h" #include "packet.h"
#include "peer.h" #include "peer.h"
@@ -36,6 +37,13 @@ struct receive {
}; };
static struct receive *receive_head = NULL; static struct receive *receive_head = NULL;
static struct flow _receive_flow = {
.name = "receive",
.new = receive_new,
.ref_count = &peer_count_in,
};
struct flow *receive_flow = &_receive_flow;
static struct parser { static struct parser {
char *name; char *name;
parser parse; parser parse;

4
adsbus/receive.h
View File

@@ -1,11 +1,11 @@
#pragma once #pragma once
#include <stdbool.h>
#define PARSER_STATE_LEN 256 #define PARSER_STATE_LEN 256
struct flow;
struct peer; struct peer;
void receive_cleanup(void); void receive_cleanup(void);
void receive_new(int, void *, struct peer *); void receive_new(int, void *, struct peer *);
void receive_print_usage(void); void receive_print_usage(void);
extern struct flow *receive_flow;

9
adsbus/send.c
View File

@@ -12,6 +12,7 @@
#include "airspy_adsb.h" #include "airspy_adsb.h"
#include "beast.h" #include "beast.h"
#include "buf.h" #include "buf.h"
#include "flow.h"
#include "json.h" #include "json.h"
#include "list.h" #include "list.h"
#include "packet.h" #include "packet.h"
@@ -32,6 +33,14 @@ struct send {
struct list_head send_list; struct list_head send_list;
}; };
static struct flow _send_flow = {
.name = "send",
.new = send_new_wrapper,
.get_hello = send_hello,
.ref_count = &peer_count_out,
};
struct flow *send_flow = &_send_flow;
typedef void (*serialize)(struct packet *, struct buf *); typedef void (*serialize)(struct packet *, struct buf *);
typedef void (*hello)(struct buf **); typedef void (*hello)(struct buf **);
static struct serializer { static struct serializer {

4
adsbus/send.h
View File

@@ -1,6 +1,9 @@
#pragma once #pragma once
#include <stdbool.h>
struct buf; struct buf;
struct flow;
struct packet; struct packet;
struct peer; struct peer;
@@ -13,3 +16,4 @@ bool send_new_hello(int, struct serializer *, struct peer *);
void send_hello(struct buf **, void *); void send_hello(struct buf **, void *);
void send_write(struct packet *); void send_write(struct packet *);
void send_print_usage(void); void send_print_usage(void);
extern struct flow *send_flow;