Factor out net services so they're not tied to a static array.

This lets different things dynamically create the services they need,
and sorts out the horrible hacks that view1090 used to make outgoing
connections. Now you can explicitly create a service and tell it to make
an outgoing connection.

This means that view1090 can now just set all the ports to zero (to disable
the listeners), do a normal net init, then explicitly construct the beast
input service without a listener and tell it to make a connection as needed.
This commit is contained in:
Oliver Jowett 2015-06-26 17:50:51 +01:00
parent 5fa039a2d4
commit 278448179d
5 changed files with 360 additions and 379 deletions

View file

@ -82,7 +82,8 @@
#include "winstubs.h" //Put everything Windows specific in here #include "winstubs.h" //Put everything Windows specific in here
#endif #endif
#include <rtl-sdr.h> // Avoid a dependency on rtl-sdr except where it's really needed.
typedef struct rtlsdr_dev rtlsdr_dev_t;
// ============================= #defines =============================== // ============================= #defines ===============================
@ -212,6 +213,7 @@
#include "util.h" #include "util.h"
#include "anet.h" #include "anet.h"
#include "net_io.h"
#include "crc.h" #include "crc.h"
#include "demod_2000.h" #include "demod_2000.h"
#include "demod_2400.h" #include "demod_2400.h"
@ -222,24 +224,6 @@
//======================== structure declarations ========================= //======================== structure declarations =========================
// Structure used to describe a networking client
struct client {
struct client* next; // Pointer to next client
int fd; // File descriptor
int service; // TCP port the client is connected to
int buflen; // Amount of data on buffer
char buf[MODES_CLIENT_BUF_SIZE+1]; // Read buffer
};
// Common writer state for all output sockets of one type
struct net_writer {
int socket; // listening socket FD, used to identify the owning service
int connections; // number of active clients
void *data; // shared write buffer, sized MODES_OUT_BUF_SIZE
int dataUsed; // number of bytes of write buffer currently used
uint64_t lastWrite; // time of last write to clients
};
// Structure representing one magnitude buffer // Structure representing one magnitude buffer
struct mag_buf { struct mag_buf {
uint16_t *data; // Magnitude data. Starts with Modes.trailing_samples worth of overlap from the previous block uint16_t *data; // Magnitude data. Starts with Modes.trailing_samples worth of overlap from the previous block
@ -287,10 +271,8 @@ struct { // Internal state
// Networking // Networking
char aneterr[ANET_ERR_LEN]; char aneterr[ANET_ERR_LEN];
struct net_service *services; // Active services
struct client *clients; // Our clients struct client *clients; // Our clients
int ris; // Raw input listening socket
int bis; // Beast input listening socket
int https; // HTTP listening socket
struct net_writer raw_out; // Raw output struct net_writer raw_out; // Raw output
struct net_writer beast_out; // Beast-format output struct net_writer beast_out; // Beast-format output
@ -448,21 +430,6 @@ void useModesMessage (struct modesMessage *mm);
// //
void interactiveShowData(void); void interactiveShowData(void);
//
// Functions exported from net_io.c
//
void modesInitNet (void);
void modesQueueOutput (struct modesMessage *mm);
void modesReadFromClient(struct client *c, char *sep, int(*handler)(struct client *, char *));
void modesNetPeriodicWork (void);
int decodeBinMessage (struct client *c, char *p);
void writeJsonToFile(const char *file, char * (*generator) (const char*,int*));
char *generateAircraftJson(const char *url_path, int *len);
char *generateReceiverJson(const char *url_path, int *len);
char *generateStatsJson(const char *url_path, int *len);
char *generateHistoryJson(const char *url_path, int *len);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

460
net_io.c
View file

@ -66,80 +66,160 @@
// function gets called and we accept new connections. All the rest is // function gets called and we accept new connections. All the rest is
// handled via non-blocking I/O and manually polling clients to see if // handled via non-blocking I/O and manually polling clients to see if
// they have something new to share with us when reading is needed. // they have something new to share with us when reading is needed.
static int decodeBinMessage(struct client *c, char *p);
static int decodeHexMessage(struct client *c, char *hex);
static int handleHTTPRequest(struct client *c, char *p);
// //
//========================================================================= //=========================================================================
// //
// Networking "stack" initialization // Networking "stack" initialization
// //
struct service {
char *descr;
int *socket;
struct net_writer *writer;
int port;
int enabled;
};
struct service services[MODES_NET_SERVICES_NUM]; // Init a service with the given read/write characteristics, return the new service.
// Doesn't arrange for the service to listen or connect
struct net_service *serviceInit(const char *descr, struct net_writer *writer, const char *sep, read_handler handler)
{
struct net_service *service;
void modesInitNet(void) { if (!(service = calloc(sizeof(*service), 1))) {
int j; fprintf(stderr, "Out of memory allocating service %s\n", descr);
exit(1);
struct service svc[MODES_NET_SERVICES_NUM] = {
{"Raw TCP output", &Modes.raw_out.socket, &Modes.raw_out, Modes.net_output_raw_port, 1},
{"Raw TCP input", &Modes.ris, NULL, Modes.net_input_raw_port, 1},
{"Beast TCP output", &Modes.beast_out.socket, &Modes.beast_out, Modes.net_output_beast_port, 1},
{"Beast TCP input", &Modes.bis, NULL, Modes.net_input_beast_port, 1},
{"HTTP server", &Modes.https, NULL, Modes.net_http_port, 1},
{"Basestation TCP output", &Modes.sbs_out.socket, &Modes.sbs_out, Modes.net_output_sbs_port, 1},
{"FlightAware TSV output", &Modes.fatsv_out.socket, &Modes.fatsv_out, Modes.net_fatsv_port, 1}
};
memcpy(&services, &svc, sizeof(svc));//services = svc;
Modes.clients = NULL;
#ifdef _WIN32
if ( (!Modes.wsaData.wVersion)
&& (!Modes.wsaData.wHighVersion) ) {
// Try to start the windows socket support
if (WSAStartup(MAKEWORD(2,1),&Modes.wsaData) != 0)
{
fprintf(stderr, "WSAStartup returned Error\n");
}
}
#endif
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
services[j].enabled = (services[j].port != 0);
if (services[j].enabled) {
int s = anetTcpServer(Modes.aneterr, services[j].port, Modes.net_bind_address);
if (s == -1) {
fprintf(stderr, "Error opening the listening port %d (%s): %s\n",
services[j].port, services[j].descr, Modes.aneterr);
exit(1);
}
anetNonBlock(Modes.aneterr, s);
*services[j].socket = s;
if (services[j].writer) {
if (! (services[j].writer->data = malloc(MODES_OUT_BUF_SIZE)) ) {
fprintf(stderr, "Out of memory allocating output buffer for service %s\n", services[j].descr);
exit(1);
}
services[j].writer->socket = s;
services[j].writer->connections = 0;
services[j].writer->dataUsed = 0;
services[j].writer->lastWrite = mstime();
}
} else {
if (Modes.debug & MODES_DEBUG_NET) printf("%s port is disabled\n", services[j].descr);
}
} }
#ifndef _WIN32 service->next = Modes.services;
Modes.services = service;
service->descr = descr;
service->listen_fd = -1;
service->connections = 0;
service->writer = writer;
service->read_sep = sep;
service->read_handler = handler;
if (service->writer) {
if (! (service->writer->data = malloc(MODES_OUT_BUF_SIZE)) ) {
fprintf(stderr, "Out of memory allocating output buffer for service %s\n", descr);
exit(1);
}
service->writer->service = service;
service->writer->dataUsed = 0;
service->writer->lastWrite = mstime();
}
return service;
}
// Create a client attached to the given service using the provided socket FD
static struct client *createClient(struct net_service *service, int fd)
{
struct client *c;
anetNonBlock(Modes.aneterr, fd);
if (!(c = (struct client *) malloc(sizeof(*c)))) {
fprintf(stderr, "Out of memory allocating a new %s network client\n", service->descr);
exit(1);
}
c->service = service;
c->next = Modes.clients;
c->fd = fd;
c->buflen = 0;
Modes.clients = c;
anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size));
++service->connections;
if (service->writer && service->connections == 1) {
service->writer->lastWrite = mstime(); // suppress heartbeat initially
}
return c;
}
// Initiate an outgoing connection which will use the given service.
// Return the new client or NULL if the connection failed
struct client *serviceConnect(struct net_service *service, char *addr, int port)
{
int s = anetTcpConnect(Modes.aneterr, addr, port);
if (s == ANET_ERR)
return NULL;
return createClient(service, s);
}
// Set up the given service to listen on an address/port.
// _exits_ on failure!
void serviceListen(struct net_service *service, char *bind_addr, int bind_port)
{
int s;
if (service->listen_fd >= 0) {
fprintf(stderr, "Tried to set up the service %s twice!\n", service->descr);
exit(1);
}
s = anetTcpServer(Modes.aneterr, bind_port, bind_addr);
if (s == ANET_ERR) {
fprintf(stderr, "Error opening the listening port %d (%s): %s\n",
bind_port, service->descr, Modes.aneterr);
exit(1);
}
anetNonBlock(Modes.aneterr, s);
service->listen_fd = s;
}
struct net_service *makeBeastInputService(void)
{
return serviceInit("Beast TCP input", NULL, NULL, decodeBinMessage);
}
void modesInitNet(void) {
struct net_service *s;
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif Modes.clients = NULL;
Modes.services = NULL;
// set up listeners
if (Modes.net_output_raw_port) {
s = serviceInit("Raw TCP output", &Modes.raw_out, NULL, NULL);
serviceListen(s, Modes.net_bind_address, Modes.net_output_raw_port);
}
if (Modes.net_output_beast_port) {
s = serviceInit("Beast TCP output", &Modes.beast_out, NULL, NULL);
serviceListen(s, Modes.net_bind_address, Modes.net_output_beast_port);
}
if (Modes.net_output_sbs_port) {
s = serviceInit("Basestation TCP output", &Modes.sbs_out, NULL, NULL);
serviceListen(s, Modes.net_bind_address, Modes.net_output_sbs_port);
}
if (Modes.net_fatsv_port) {
s = serviceInit("FATSV TCP output", &Modes.fatsv_out, NULL, NULL);
serviceListen(s, Modes.net_bind_address, Modes.net_fatsv_port);
}
if (Modes.net_input_raw_port) {
s = serviceInit("Raw TCP input", NULL, "\n", decodeHexMessage);
serviceListen(s, Modes.net_bind_address, Modes.net_input_raw_port);
}
if (Modes.net_input_beast_port) {
s = makeBeastInputService();
serviceListen(s, Modes.net_bind_address, Modes.net_input_beast_port);
}
if (Modes.net_http_port) {
s = serviceInit("HTTP server", NULL, "\r\n\r\n", handleHTTPRequest);
serviceListen(s, Modes.net_bind_address, Modes.net_http_port);
}
} }
// //
//========================================================================= //=========================================================================
@ -147,37 +227,18 @@ void modesInitNet(void) {
// This function gets called from time to time when the decoding thread is // This function gets called from time to time when the decoding thread is
// awakened by new data arriving. This usually happens a few times every second // awakened by new data arriving. This usually happens a few times every second
// //
struct client * modesAcceptClients(void) { static struct client * modesAcceptClients(void) {
int fd, port; int fd, port;
unsigned int j; struct net_service *s;
struct client *c;
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) { for (s = Modes.services; s; s = s->next) {
if (services[j].enabled) { if (s->listen_fd >= 0) {
fd = anetTcpAccept(Modes.aneterr, *services[j].socket, NULL, &port); while ((fd = anetTcpAccept(Modes.aneterr, s->listen_fd, NULL, &port)) >= 0) {
if (fd == -1) continue; createClient(s, fd);
}
anetNonBlock(Modes.aneterr, fd); }
c = (struct client *) malloc(sizeof(*c));
c->service = *services[j].socket;
c->next = Modes.clients;
c->fd = fd;
c->buflen = 0;
Modes.clients = c;
anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size));
if (services[j].writer) {
if (++ services[j].writer->connections == 1) {
services[j].writer->lastWrite = mstime(); // suppress heartbeat initially
}
}
j--; // Try again with the same listening port
if (Modes.debug & MODES_DEBUG_NET)
printf("Created new client %d\n", fd);
}
} }
return Modes.clients; return Modes.clients;
} }
// //
@ -185,8 +246,11 @@ struct client * modesAcceptClients(void) {
// //
// On error free the client, collect the structure, adjust maxfd if needed. // On error free the client, collect the structure, adjust maxfd if needed.
// //
void modesCloseClient(struct client *c) { static void modesCloseClient(struct client *c) {
int j; if (!c->service) {
fprintf(stderr, "warning: double close of net client\n");
return;
}
// Clean up, but defer removing from the list until modesNetCleanup(). // Clean up, but defer removing from the list until modesNetCleanup().
// This is because there may be stackframes still pointing at this // This is because there may be stackframes still pointing at this
@ -194,21 +258,11 @@ void modesCloseClient(struct client *c) {
// be freed) // be freed)
close(c->fd); close(c->fd);
c->service->connections--;
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
if (c->service == *services[j].socket) {
if (services[j].writer)
services[j].writer->connections--;
break;
}
}
if (Modes.debug & MODES_DEBUG_NET)
printf("Closing client %d\n", c->fd);
// mark it as inactive and ready to be freed // mark it as inactive and ready to be freed
c->fd = -1; c->fd = -1;
c->service = -1; c->service = NULL;
} }
// //
//========================================================================= //=========================================================================
@ -219,7 +273,7 @@ static void flushWrites(struct net_writer *writer) {
struct client *c; struct client *c;
for (c = Modes.clients; c; c = c->next) { for (c = Modes.clients; c; c = c->next) {
if (c->service == writer->socket) { if (c->service == writer->service) {
#ifndef _WIN32 #ifndef _WIN32
int nwritten = write(c->fd, writer->data, writer->dataUsed); int nwritten = write(c->fd, writer->data, writer->dataUsed);
#else #else
@ -238,31 +292,32 @@ static void flushWrites(struct net_writer *writer) {
// Prepare to write up to 'len' bytes to the given net_writer. // Prepare to write up to 'len' bytes to the given net_writer.
// Returns a pointer to write to, or NULL to skip this write. // Returns a pointer to write to, or NULL to skip this write.
static void *prepareWrite(struct net_writer *writer, int len) { static void *prepareWrite(struct net_writer *writer, int len) {
if (!writer || if (!writer ||
!writer->connections || !writer->service ||
!writer->data) !writer->service->connections ||
return NULL; !writer->data)
return NULL;
if (len > MODES_OUT_BUF_SIZE) if (len > MODES_OUT_BUF_SIZE)
return NULL; return NULL;
if (writer->dataUsed + len >= MODES_OUT_BUF_SIZE) { if (writer->dataUsed + len >= MODES_OUT_BUF_SIZE) {
// Flush now to free some space // Flush now to free some space
flushWrites(writer); flushWrites(writer);
} }
return writer->data + writer->dataUsed; return writer->data + writer->dataUsed;
} }
// Complete a write previously begun by prepareWrite. // Complete a write previously begun by prepareWrite.
// endptr should point one byte past the last byte written // endptr should point one byte past the last byte written
// to the buffer returned from prepareWrite. // to the buffer returned from prepareWrite.
static void completeWrite(struct net_writer *writer, void *endptr) { static void completeWrite(struct net_writer *writer, void *endptr) {
writer->dataUsed = endptr - writer->data; writer->dataUsed = endptr - writer->data;
if (writer->dataUsed >= Modes.net_output_flush_size) { if (writer->dataUsed >= Modes.net_output_flush_size) {
flushWrites(writer); flushWrites(writer);
} }
} }
// //
@ -270,7 +325,7 @@ static void completeWrite(struct net_writer *writer, void *endptr) {
// //
// Write raw output in Beast Binary format with Timestamp to TCP clients // Write raw output in Beast Binary format with Timestamp to TCP clients
// //
void modesSendBeastOutput(struct modesMessage *mm) { static void modesSendBeastOutput(struct modesMessage *mm) {
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.beast_out, 2 + 2 * (7 + msgLen)); char *p = prepareWrite(&Modes.beast_out, 2 + 2 * (7 + msgLen));
char ch; char ch;
@ -320,7 +375,7 @@ void modesSendBeastOutput(struct modesMessage *mm) {
// //
// Write raw output to TCP clients // Write raw output to TCP clients
// //
void modesSendRawOutput(struct modesMessage *mm) { static void modesSendRawOutput(struct modesMessage *mm) {
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.raw_out, msgLen*2 + 15); char *p = prepareWrite(&Modes.raw_out, msgLen*2 + 15);
int j; int j;
@ -353,7 +408,7 @@ void modesSendRawOutput(struct modesMessage *mm) {
// Write SBS output to TCP clients // Write SBS output to TCP clients
// The message structure mm->bFlags tells us what has been updated by this message // The message structure mm->bFlags tells us what has been updated by this message
// //
void modesSendSBSOutput(struct modesMessage *mm) { static void modesSendSBSOutput(struct modesMessage *mm) {
char *p; char *p;
struct timespec now; struct timespec now;
struct tm stTime_receive, stTime_now; struct tm stTime_receive, stTime_now;
@ -536,7 +591,7 @@ void modesQueueOutput(struct modesMessage *mm) {
// The function always returns 0 (success) to the caller as there is no // The function always returns 0 (success) to the caller as there is no
// case where we want broken messages here to close the client connection. // case where we want broken messages here to close the client connection.
// //
int decodeBinMessage(struct client *c, char *p) { static int decodeBinMessage(struct client *c, char *p) {
int msgLen = 0; int msgLen = 0;
int j; int j;
char ch; char ch;
@ -611,7 +666,7 @@ int decodeBinMessage(struct client *c, char *p) {
// Turn an hex digit into its 4 bit decimal value. // Turn an hex digit into its 4 bit decimal value.
// Returns -1 if the digit is not in the 0-F range. // Returns -1 if the digit is not in the 0-F range.
// //
int hexDigitVal(int c) { static int hexDigitVal(int c) {
c = tolower(c); c = tolower(c);
if (c >= '0' && c <= '9') return c-'0'; if (c >= '0' && c <= '9') return c-'0';
else if (c >= 'a' && c <= 'f') return c-'a'+10; else if (c >= 'a' && c <= 'f') return c-'a'+10;
@ -631,7 +686,7 @@ int hexDigitVal(int c) {
// The function always returns 0 (success) to the caller as there is no // The function always returns 0 (success) to the caller as there is no
// case where we want broken messages here to close the client connection. // case where we want broken messages here to close the client connection.
// //
int decodeHexMessage(struct client *c, char *hex) { static int decodeHexMessage(struct client *c, char *hex) {
int l = strlen(hex), j; int l = strlen(hex), j;
unsigned char msg[MODES_LONG_MSG_BYTES]; unsigned char msg[MODES_LONG_MSG_BYTES];
struct modesMessage mm; struct modesMessage mm;
@ -1102,7 +1157,7 @@ static struct {
// Returns 1 on error to signal the caller the client connection should // Returns 1 on error to signal the caller the client connection should
// be closed. // be closed.
// //
int handleHTTPRequest(struct client *c, char *p) { static int handleHTTPRequest(struct client *c, char *p) {
char hdr[512]; char hdr[512];
int clen, hdrlen; int clen, hdrlen;
int httpver, keepalive; int httpver, keepalive;
@ -1277,8 +1332,7 @@ int handleHTTPRequest(struct client *c, char *p) {
// The handler returns 0 on success, or 1 to signal this function we should // The handler returns 0 on success, or 1 to signal this function we should
// close the connection with the client in case of non-recoverable errors. // close the connection with the client in case of non-recoverable errors.
// //
void modesReadFromClient(struct client *c, char *sep, static void modesReadFromClient(struct client *c) {
int(*handler)(struct client *, char *)) {
int left; int left;
int nread; int nread;
int fullmsg; int fullmsg;
@ -1332,7 +1386,7 @@ void modesReadFromClient(struct client *c, char *sep,
e = s = c->buf; // Start with the start of buffer, first message e = s = c->buf; // Start with the start of buffer, first message
if (c->service == Modes.bis) { if (c->service->read_sep == NULL) {
// This is the Beast Binary scanning case. // This is the Beast Binary scanning case.
// If there is a complete message still in the buffer, there must be the separator 'sep' // If there is a complete message still in the buffer, there must be the separator 'sep'
// in the buffer, note that we full-scan the buffer at every read for simplicity. // in the buffer, note that we full-scan the buffer at every read for simplicity.
@ -1366,7 +1420,7 @@ void modesReadFromClient(struct client *c, char *sep,
break; break;
} }
// Have a 0x1a followed by 1, 2 or 3 - pass message less 0x1a to handler. // Have a 0x1a followed by 1, 2 or 3 - pass message less 0x1a to handler.
if (handler(c, s)) { if (c->service->read_handler(c, s)) {
modesCloseClient(c); modesCloseClient(c);
return; return;
} }
@ -1380,13 +1434,13 @@ void modesReadFromClient(struct client *c, char *sep,
// If there is a complete message still in the buffer, there must be the separator 'sep' // If there is a complete message still in the buffer, there must be the separator 'sep'
// in the buffer, note that we full-scan the buffer at every read for simplicity. // in the buffer, note that we full-scan the buffer at every read for simplicity.
// //
while ((e = strstr(s, sep)) != NULL) { // end of first message if found while ((e = strstr(s, c->service->read_sep)) != NULL) { // end of first message if found
*e = '\0'; // The handler expects null terminated strings *e = '\0'; // The handler expects null terminated strings
if (handler(c, s)) { // Pass message to handler. if (c->service->read_handler(c, s)) { // Pass message to handler.
modesCloseClient(c); // Handler returns 1 on error to signal we . modesCloseClient(c); // Handler returns 1 on error to signal we .
return; // should close the client connection return; // should close the client connection
} }
s = e + strlen(sep); // Move to start of next message s = e + strlen(c->service->read_sep); // Move to start of next message
fullmsg = 1; fullmsg = 1;
} }
} }
@ -1402,13 +1456,14 @@ void modesReadFromClient(struct client *c, char *sep,
#define TSV_MAX_PACKET_SIZE 160 #define TSV_MAX_PACKET_SIZE 160
static void writeFATSV() { static void writeFATSV()
{
struct aircraft *a; struct aircraft *a;
uint64_t now; uint64_t now;
static uint64_t next_update; static uint64_t next_update;
if (!Modes.fatsv_out.connections) { if (!Modes.fatsv_out.service || !Modes.fatsv_out.service->connections) {
return; // no active connections return; // not enabled or no active connections
} }
now = mstime(); now = mstime();
@ -1595,82 +1650,77 @@ static void writeFATSV() {
// Perform periodic network work // Perform periodic network work
// //
void modesNetPeriodicWork(void) { void modesNetPeriodicWork(void) {
struct client *c, **prev; struct client *c, **prev;
uint64_t now = mstime(); struct net_service *s;
int j; uint64_t now = mstime();
int need_heartbeat = 0, need_flush = 0; int need_heartbeat = 0, need_flush = 0;
// Accept new connetions // Accept new connetions
modesAcceptClients(); modesAcceptClients();
// Read from clients // Read from clients
for (c = Modes.clients; c; c = c->next) { for (c = Modes.clients; c; c = c->next) {
if (c->service == Modes.ris) { if (c->service->read_handler)
modesReadFromClient(c,"\n",decodeHexMessage); modesReadFromClient(c);
} else if (c->service == Modes.bis) { }
modesReadFromClient(c,"",decodeBinMessage);
} else if (c->service == Modes.https) {
modesReadFromClient(c,"\r\n\r\n",handleHTTPRequest);
}
}
// Generate FATSV output // Generate FATSV output
writeFATSV(); writeFATSV();
// If we have generated no messages for a while, generate // If we have generated no messages for a while, generate
// a dummy heartbeat message. // a dummy heartbeat message.
if (Modes.net_heartbeat_interval) { if (Modes.net_heartbeat_interval) {
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) { for (s = Modes.services; s; s = s->next) {
if (services[j].writer && if (s->writer &&
services[j].writer->connections && s->connections &&
(services[j].writer->lastWrite + Modes.net_heartbeat_interval) <= now) { (s->writer->lastWrite + Modes.net_heartbeat_interval) <= now) {
need_flush = 1; need_flush = 1;
if (services[j].writer->dataUsed == 0) { if (s->writer->dataUsed == 0) {
need_heartbeat = 1; need_heartbeat = 1;
break; break;
} }
} }
}
} }
}
if (need_heartbeat) { if (need_heartbeat) {
// //
// We haven't sent any traffic for some time. To try and keep any TCP // We haven't sent any traffic for some time. To try and keep any TCP
// links alive, send a null frame. This will help stop any routers discarding our TCP // links alive, send a null frame. This will help stop any routers discarding our TCP
// link which will cause an un-recoverable link error if/when a real frame arrives. // link which will cause an un-recoverable link error if/when a real frame arrives.
// //
// Fudge up a null message // Fudge up a null message
struct modesMessage mm; struct modesMessage mm;
memset(&mm, 0, sizeof(mm)); memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS; mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = 0; mm.timestampMsg = 0;
mm.msgtype = -1; mm.msgtype = -1;
// Feed output clients // Feed output clients
modesQueueOutput(&mm); modesQueueOutput(&mm);
}
// If we have data that has been waiting to be written for a while,
// write it now.
for (s = Modes.services; s; s = s->next) {
if (s->writer &&
s->writer->dataUsed &&
(need_flush || (s->writer->lastWrite + Modes.net_output_flush_interval) <= now)) {
flushWrites(s->writer);
} }
}
// If we have data that has been waiting to be written for a while, // Unlink and free closed clients
// write it now. for (prev = &Modes.clients, c = *prev; c; c = *prev) {
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) { if (c->fd == -1) {
if (services[j].writer && // Recently closed, prune from list
services[j].writer->dataUsed && *prev = c->next;
(need_flush || (services[j].writer->lastWrite + Modes.net_output_flush_interval) <= now)) { free(c);
flushWrites(services[j].writer); } else {
} prev = &c->next;
} }
}
// Unlink and free closed clients
for (prev = &Modes.clients, c = *prev; c; c = *prev) {
if (c->fd == -1) {
// Recently closed, prune from list
*prev = c->next;
free(c);
} else {
prev = &c->next;
}
}
} }
// //

78
net_io.h Normal file
View file

@ -0,0 +1,78 @@
// Part of dump1090, a Mode S message decoder for RTLSDR devices.
//
// net_io.h: network handling.
//
// Copyright (c) 2014,2015 Oliver Jowett <oliver@mutability.co.uk>
//
// This file is free software: you may copy, redistribute and/or modify it
// under the terms of the GNU General Public License as published by the
// Free Software Foundation, either version 2 of the License, or (at your
// option) any later version.
//
// This file is distributed in the hope that it will be useful, but
// WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#ifndef DUMP1090_NETIO_H
#define DUMP1090_NETIO_H
// Describes a networking service (group of connections)
struct modesMessage;
struct client;
typedef int (*read_handler)(struct client *, char *);
// Describes one network service (a group of clients with common behaviour)
struct net_service {
struct net_service* next;
const char *descr;
int listen_fd;
int connections; // number of active clients
struct net_writer *writer; // shared writer state
const char *read_sep; // hander details for input data
read_handler read_handler;
};
// Structure used to describe a networking client
struct client {
struct client* next; // Pointer to next client
int fd; // File descriptor
struct net_service *service; // Service this client is part of
int buflen; // Amount of data on buffer
char buf[MODES_CLIENT_BUF_SIZE+1]; // Read buffer
};
// Common writer state for all output sockets of one type
struct net_writer {
struct net_service *service; // owning service
void *data; // shared write buffer, sized MODES_OUT_BUF_SIZE
int dataUsed; // number of bytes of write buffer currently used
uint64_t lastWrite; // time of last write to clients
};
struct net_service *serviceInit(const char *descr, struct net_writer *writer, const char *sep, read_handler handler);
struct client *serviceConnect(struct net_service *service, char *addr, int port);
void serviceListen(struct net_service *service, char *bind_addr, int bind_port);
// view1090 / faup1090 want to create this themselves:
struct net_service *makeBeastInputService(void);
void modesInitNet(void);
void modesQueueOutput(struct modesMessage *mm);
void modesNetPeriodicWork(void);
// TODO: move these somewhere else
char *generateAircraftJson(const char *url_path, int *len);
char *generateStatsJson(const char *url_path, int *len);
char *generateReceiverJson(const char *url_path, int *len);
char *generateHistoryJson(const char *url_path, int *len);
void writeJsonToFile(const char *file, char * (*generator) (const char *,int*));
#endif

View file

@ -27,12 +27,12 @@
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
// //
#include "view1090.h" #include "dump1090.h"
// //
// ============================= Utility functions ========================== // ============================= Utility functions ==========================
// //
void sigintHandler(int dummy) { void sigintHandler(int dummy) {
NOTUSED(dummy); MODES_NOTUSED(dummy);
signal(SIGINT, SIG_DFL); // reset signal handler - bit extra safety signal(SIGINT, SIG_DFL); // reset signal handler - bit extra safety
Modes.exit = 1; // Signal to threads that we are done Modes.exit = 1; // Signal to threads that we are done
} }
@ -63,15 +63,11 @@ int getTermRows() { return MODES_INTERACTIVE_ROWS;}
void view1090InitConfig(void) { void view1090InitConfig(void) {
// Default everything to zero/NULL // Default everything to zero/NULL
memset(&Modes, 0, sizeof(Modes)); memset(&Modes, 0, sizeof(Modes));
memset(&View1090, 0, sizeof(View1090));
// Now initialise things that should not be 0/NULL to their defaults // Now initialise things that should not be 0/NULL to their defaults
Modes.check_crc = 1; Modes.check_crc = 1;
strcpy(View1090.net_input_beast_ipaddr,VIEW1090_NET_OUTPUT_IP_ADDRESS);
Modes.net_input_beast_port = MODES_NET_OUTPUT_BEAST_PORT;
Modes.interactive_rows = getTermRows(); Modes.interactive_rows = getTermRows();
Modes.interactive_display_ttl = MODES_INTERACTIVE_DISPLAY_TTL; Modes.interactive_display_ttl = MODES_INTERACTIVE_DISPLAY_TTL;
Modes.interactive = 1; Modes.interactive = 1;
} }
// //
@ -117,33 +113,6 @@ void view1090Init(void) {
icaoFilterInit(); icaoFilterInit();
} }
// Set up data connection
int setupConnection(struct client *c) {
int fd;
// Try to connect to the selected ip address and port. We only support *ONE* input connection which we initiate.here.
if ((fd = anetTcpConnect(Modes.aneterr, View1090.net_input_beast_ipaddr, Modes.net_input_beast_port)) != ANET_ERR) {
anetNonBlock(Modes.aneterr, fd);
//
// Setup a service callback client structure for a beast binary input (from dump1090)
// This is a bit dodgy under Windows. The fd parameter is a handle to the internet
// socket on which we are receiving data. Under Linux, these seem to start at 0 and
// count upwards. However, Windows uses "HANDLES" and these don't nececeriy start at 0.
// dump1090 limits fd to values less than 1024, and then uses the fd parameter to
// index into an array of clients. This is ok-ish if handles are allocated up from 0.
// However, there is no gaurantee that Windows will behave like this, and if Windows
// allocates a handle greater than 1024, then dump1090 won't like it. On my test machine,
// the first Windows handle is usually in the 0x54 (84 decimal) region.
c->next = NULL;
c->buflen = 0;
c->fd =
c->service =
Modes.bis = fd;
Modes.clients = c;
}
return fd;
}
// //
// ================================ Main ==================================== // ================================ Main ====================================
// //
@ -175,9 +144,11 @@ void showHelp(void) {
//========================================================================= //=========================================================================
// //
int main(int argc, char **argv) { int main(int argc, char **argv) {
int j, fd; int j;
struct client *c; struct client *c;
char pk_buf[8]; struct net_service *s;
char *bo_connect_ipaddr = "127.0.0.1";
int bo_connect_port = MODES_NET_OUTPUT_BEAST_PORT;
// Set sane defaults // Set sane defaults
@ -189,9 +160,9 @@ int main(int argc, char **argv) {
int more = ((j + 1) < argc); // There are more arguments int more = ((j + 1) < argc); // There are more arguments
if (!strcmp(argv[j],"--net-bo-port") && more) { if (!strcmp(argv[j],"--net-bo-port") && more) {
Modes.net_input_beast_port = atoi(argv[++j]); bo_connect_port = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-bo-ipaddr") && more) { } else if (!strcmp(argv[j],"--net-bo-ipaddr") && more) {
strcpy(View1090.net_input_beast_ipaddr, argv[++j]); bo_connect_ipaddr = argv[++j];
} else if (!strcmp(argv[j],"--modeac")) { } else if (!strcmp(argv[j],"--modeac")) {
Modes.mode_ac = 1; Modes.mode_ac = 1;
} else if (!strcmp(argv[j],"--interactive-rows") && more) { } else if (!strcmp(argv[j],"--interactive-rows") && more) {
@ -240,11 +211,13 @@ int main(int argc, char **argv) {
// Initialization // Initialization
view1090Init(); view1090Init();
modesInitNet();
// Try to connect to the selected ip address and port. We only support *ONE* input connection which we initiate.here. // Try to connect to the selected ip address and port. We only support *ONE* input connection which we initiate.here.
c = (struct client *) malloc(sizeof(*c)); s = makeBeastInputService();
if ((fd = setupConnection(c)) == ANET_ERR) { c = serviceConnect(s, bo_connect_ipaddr, bo_connect_port);
fprintf(stderr, "Failed to connect to %s:%d\n", View1090.net_input_beast_ipaddr, Modes.net_input_beast_port); if (!c) {
fprintf(stderr, "Failed to connect to %s:%d: %s\n", bo_connect_ipaddr, bo_connect_port, Modes.aneterr);
exit(1); exit(1);
} }
@ -252,21 +225,18 @@ int main(int argc, char **argv) {
while (!Modes.exit) { while (!Modes.exit) {
icaoFilterExpire(); icaoFilterExpire();
trackPeriodicUpdate(); trackPeriodicUpdate();
modesNetPeriodicWork();
interactiveShowData(); interactiveShowData();
if ((fd == ANET_ERR) || (recv(c->fd, pk_buf, sizeof(pk_buf), MSG_PEEK | MSG_DONTWAIT) == 0)) {
free(c);
usleep(1000000);
c = (struct client *) malloc(sizeof(*c));
fd = setupConnection(c);
continue;
}
modesReadFromClient(c,"",decodeBinMessage);
usleep(100000);
}
// The user has stopped us, so close any socket we opened if (s->connections == 0) {
if (fd != ANET_ERR) // lost input connection, try to reconnect
{close(fd);} usleep(1000000);
c = serviceConnect(s, bo_connect_ipaddr, bo_connect_port);
continue;
}
usleep(100000);
}
return (0); return (0);
} }

View file

@ -1,84 +0,0 @@
// view1090, a Mode S messages viewer for dump1090 devices.
//
// Copyright (C) 2013 by Malcolm Robb <Support@ATTAvionics.com>
//
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// * Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
#ifndef __VIEW1090_H
#define __VIEW1090_H
// ============================= Include files ==========================
#include "dump1090.h"
#ifndef _WIN32
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdint.h>
#include <errno.h>
#include <unistd.h>
#include <math.h>
#include <sys/time.h>
#include <sys/timeb.h>
#include <signal.h>
#include <fcntl.h>
#include <ctype.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/socket.h>
#include "rtl-sdr.h"
#include "anet.h"
#else
#include "winstubs.h" //Put everything Windows specific in here
#endif
// ============================= #defines ===============================
#define VIEW1090_NET_OUTPUT_IP_ADDRESS "127.0.0.1"
#define NOTUSED(V) ((void) V)
// ======================== structure declarations ========================
// Program global state
struct { // Internal state
// Networking
char net_input_beast_ipaddr[32]; // IPv4 address or network name of server/RPi
} View1090;
// ======================== function declarations =========================
#ifdef __cplusplus
extern "C" {
#endif
#ifdef __cplusplus
}
#endif
#endif // __VIEW1090_H