AFG/dump1090
3
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Merge branch 'net-cleanups' into oversampling

Browse source 
Conflicts:
	dump1090.c
	dump1090.h
This commit is contained in:
Oliver Jowett 2014-10-03 23:04:09 +01:00
parent 4a4fed2b37 1769ac9006
commit 22f2a8db37
4 changed files with 216 additions and 197 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
dump1090.c
View file

@ -70,7 +70,7 @@ void modesInitConfig(void) {
Modes.freq = MODES_DEFAULT_FREQ; Modes.freq = MODES_DEFAULT_FREQ;
Modes.ppm_error = MODES_DEFAULT_PPM; Modes.ppm_error = MODES_DEFAULT_PPM;
Modes.check_crc = 1; Modes.check_crc = 1;
Modes.net_heartbeat_rate = MODES_NET_HEARTBEAT_RATE; Modes.net_heartbeat_interval = MODES_NET_HEARTBEAT_INTERVAL;
Modes.net_output_sbs_port = MODES_NET_OUTPUT_SBS_PORT; Modes.net_output_sbs_port = MODES_NET_OUTPUT_SBS_PORT;
Modes.net_output_raw_port = MODES_NET_OUTPUT_RAW_PORT; Modes.net_output_raw_port = MODES_NET_OUTPUT_RAW_PORT;
Modes.net_input_raw_port = MODES_NET_INPUT_RAW_PORT; Modes.net_input_raw_port = MODES_NET_INPUT_RAW_PORT;
@ -99,9 +99,7 @@ void modesInit(void) {
((Modes.pFileData = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE) ) == NULL) || ((Modes.pFileData = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE) ) == NULL) ||
((Modes.magnitude = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE+Modes.trailing_space) ) == NULL) || ((Modes.magnitude = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE+Modes.trailing_space) ) == NULL) ||
((Modes.maglut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) || ((Modes.maglut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) ||
((Modes.log10lut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) || ((Modes.log10lut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) )
((Modes.beastOut = (char *) malloc(MODES_RAWOUT_BUF_SIZE) ) == NULL) ||
((Modes.rawOut = (char *) malloc(MODES_RAWOUT_BUF_SIZE) ) == NULL) )
{ {
fprintf(stderr, "Out of memory allocating data buffer.\n"); fprintf(stderr, "Out of memory allocating data buffer.\n");
exit(1); exit(1);
@ -132,10 +130,10 @@ void modesInit(void) {
} }
// Limit the maximum requested raw output size to less than one Ethernet Block // Limit the maximum requested raw output size to less than one Ethernet Block
if (Modes.net_output_raw_size > (MODES_RAWOUT_BUF_FLUSH)) if (Modes.net_output_flush_size > (MODES_OUT_FLUSH_SIZE))
{Modes.net_output_raw_size = MODES_RAWOUT_BUF_FLUSH;} {Modes.net_output_flush_size = MODES_OUT_FLUSH_SIZE;}
if (Modes.net_output_raw_rate > (MODES_RAWOUT_BUF_RATE)) if (Modes.net_output_flush_interval > (MODES_OUT_FLUSH_INTERVAL))
{Modes.net_output_raw_rate = MODES_RAWOUT_BUF_RATE;} {Modes.net_output_flush_interval = MODES_OUT_FLUSH_INTERVAL;}
if (Modes.net_sndbuf_size > (MODES_NET_SNDBUF_MAX)) if (Modes.net_sndbuf_size > (MODES_NET_SNDBUF_MAX))
{Modes.net_sndbuf_size = MODES_NET_SNDBUF_MAX;} {Modes.net_sndbuf_size = MODES_NET_SNDBUF_MAX;}
@ -417,7 +415,6 @@ void showHelp(void) {
"--raw Show only messages hex values\n" "--raw Show only messages hex values\n"
"--net Enable networking\n" "--net Enable networking\n"
"--modeac Enable decoding of SSR Modes 3/A & 3/C\n" "--modeac Enable decoding of SSR Modes 3/A & 3/C\n"
"--net-beast TCP raw output in Beast binary format\n"
"--net-only Enable just networking, no RTL device or file used\n" "--net-only Enable just networking, no RTL device or file used\n"
"--net-http-port <port> HTTP server port (default: 8080)\n" "--net-http-port <port> HTTP server port (default: 8080)\n"
"--net-ri-port <port> TCP raw input listen port (default: 30001)\n" "--net-ri-port <port> TCP raw input listen port (default: 30001)\n"
@ -425,8 +422,8 @@ void showHelp(void) {
"--net-sbs-port <port> TCP BaseStation output listen port (default: 30003)\n" "--net-sbs-port <port> TCP BaseStation output listen port (default: 30003)\n"
"--net-bi-port <port> TCP Beast input listen port (default: 30004)\n" "--net-bi-port <port> TCP Beast input listen port (default: 30004)\n"
"--net-bo-port <port> TCP Beast output listen port (default: 30005)\n" "--net-bo-port <port> TCP Beast output listen port (default: 30005)\n"
"--net-ro-size <size> TCP raw output minimum size (default: 0)\n" "--net-ro-size <size> TCP output minimum size (default: 0)\n"
"--net-ro-rate <rate> TCP raw output memory flush rate (default: 0)\n" "--net-ro-interval <rate> TCP output memory flush rate in seconds (default: 0)\n"
"--net-heartbeat <rate> TCP heartbeat rate in seconds (default: 60 sec; 0 to disable)\n" "--net-heartbeat <rate> TCP heartbeat rate in seconds (default: 60 sec; 0 to disable)\n"
"--net-buffer <n> TCP buffer size 64Kb * (2^n) (default: n=0, 64Kb)\n" "--net-buffer <n> TCP buffer size 64Kb * (2^n) (default: n=0, 64Kb)\n"
"--lat <latitude> Reference/receiver latitude for surface posn (opt)\n" "--lat <latitude> Reference/receiver latitude for surface posn (opt)\n"
@ -607,8 +604,7 @@ void backgroundTasks(void) {
static time_t next_stats; static time_t next_stats;
if (Modes.net) { if (Modes.net) {
modesReadFromClients(); modesNetPeriodicWork();
modesNetCleanup();
} }
// If Modes.aircrafts is not NULL, remove any stale aircraft // If Modes.aircrafts is not NULL, remove any stale aircraft
@ -736,11 +732,13 @@ int main(int argc, char **argv) {
Modes.net = 1; Modes.net = 1;
Modes.net_only = 1; Modes.net_only = 1;
} else if (!strcmp(argv[j],"--net-heartbeat") && more) { } else if (!strcmp(argv[j],"--net-heartbeat") && more) {
Modes.net_heartbeat_rate = atoi(argv[++j]) * 15; Modes.net_heartbeat_interval = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-size") && more) { } else if (!strcmp(argv[j],"--net-ro-size") && more) {
Modes.net_output_raw_size = atoi(argv[++j]); Modes.net_output_flush_size = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-rate") && more) { } else if (!strcmp(argv[j],"--net-ro-rate") && more) {
Modes.net_output_raw_rate = atoi(argv[++j]); Modes.net_output_flush_interval = atoi(argv[++j]) / 15; // backwards compatibility
} else if (!strcmp(argv[j],"--net-ro-interval") && more) {
Modes.net_output_flush_interval = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-port") && more) { } else if (!strcmp(argv[j],"--net-ro-port") && more) {
if (Modes.beast) // Required for legacy backward compatibility if (Modes.beast) // Required for legacy backward compatibility
{Modes.net_output_beast_port = atoi(argv[++j]);;} {Modes.net_output_beast_port = atoi(argv[++j]);;}

44
dump1090.h
View file

@ -127,9 +127,9 @@
#define MODES_OS_LONG_MSG_SIZE (MODES_LONG_MSG_SAMPLES * sizeof(uint16_t)) #define MODES_OS_LONG_MSG_SIZE (MODES_LONG_MSG_SAMPLES * sizeof(uint16_t))
#define MODES_OS_SHORT_MSG_SIZE (MODES_SHORT_MSG_SAMPLES * sizeof(uint16_t)) #define MODES_OS_SHORT_MSG_SIZE (MODES_SHORT_MSG_SAMPLES * sizeof(uint16_t))
#define MODES_RAWOUT_BUF_SIZE (1500) #define MODES_OUT_BUF_SIZE (1500)
#define MODES_RAWOUT_BUF_FLUSH (MODES_RAWOUT_BUF_SIZE - 200) #define MODES_OUT_FLUSH_SIZE (MODES_OUT_BUF_SIZE - 256)
#define MODES_RAWOUT_BUF_RATE (1000) // 1000 * 64mS = 1 Min approx #define MODES_OUT_FLUSH_INTERVAL (60)
#define MODES_ICAO_CACHE_LEN 1024 // Power of two required #define MODES_ICAO_CACHE_LEN 1024 // Power of two required
#define MODES_ICAO_CACHE_TTL 60 // Time to live of cached addresses #define MODES_ICAO_CACHE_TTL 60 // Time to live of cached addresses
@ -176,7 +176,7 @@
#define MODES_INTERACTIVE_DELETE_TTL 300 // Delete from the list after 300 seconds #define MODES_INTERACTIVE_DELETE_TTL 300 // Delete from the list after 300 seconds
#define MODES_INTERACTIVE_DISPLAY_TTL 60 // Delete from display after 60 seconds #define MODES_INTERACTIVE_DISPLAY_TTL 60 // Delete from display after 60 seconds
#define MODES_NET_HEARTBEAT_RATE 900 // Each block is approx 65mS - default is > 1 min #define MODES_NET_HEARTBEAT_INTERVAL 60 // seconds
#define MODES_NET_SERVICES_NUM 6 #define MODES_NET_SERVICES_NUM 6
#define MODES_NET_INPUT_RAW_PORT 30001 #define MODES_NET_INPUT_RAW_PORT 30001
@ -264,6 +264,15 @@ struct demod_stats {
unsigned int bit_fix[MODES_MAX_BITERRORS]; unsigned int bit_fix[MODES_MAX_BITERRORS];
}; };
// 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
time_t lastWrite; // time of last write to clients
};
// Program global state // Program global state
struct { // Internal state struct { // Internal state
pthread_t reader_thread; pthread_t reader_thread;
@ -300,16 +309,14 @@ struct { // Internal state
// Networking // Networking
char aneterr[ANET_ERR_LEN]; char aneterr[ANET_ERR_LEN];
struct client *clients; // Our clients struct client *clients; // Our clients
int sbsos; // SBS output listening socket
int ros; // Raw output listening socket
int ris; // Raw input listening socket int ris; // Raw input listening socket
int bos; // Beast output listening socket
int bis; // Beast input listening socket int bis; // Beast input listening socket
int https; // HTTP listening socket int https; // HTTP listening socket
char *rawOut; // Buffer for building raw output data
int rawOutUsed; // How much of the buffer is currently used struct net_writer raw_out; // Raw output
char *beastOut; // Buffer for building beast output data struct net_writer beast_out; // Beast-format output
int beastOutUsed; // How much if the buffer is currently used struct net_writer sbs_out; // SBS-format output
#ifdef _WIN32 #ifdef _WIN32
WSADATA wsaData; // Windows socket initialisation WSADATA wsaData; // Windows socket initialisation
#endif #endif
@ -326,12 +333,10 @@ struct { // Internal state
int debug; // Debugging mode int debug; // Debugging mode
int net; // Enable networking int net; // Enable networking
int net_only; // Enable just networking int net_only; // Enable just networking
int net_heartbeat_count; // TCP heartbeat counter int net_heartbeat_interval; // TCP heartbeat interval (seconds)
int net_heartbeat_rate; // TCP heartbeat rate
int net_output_sbs_port; // SBS output TCP port int net_output_sbs_port; // SBS output TCP port
int net_output_raw_size; // Minimum Size of the output raw data int net_output_flush_size; // Minimum Size of output data
int net_output_raw_rate; // Rate (in 64mS increments) of output raw data int net_output_flush_interval; // Maximum interval (in seconds) between outputwrites
int net_output_raw_rate_count; // Rate (in 64mS increments) of output raw data
int net_output_raw_port; // Raw output TCP port int net_output_raw_port; // Raw output TCP port
int net_input_raw_port; // Raw input TCP port int net_input_raw_port; // Raw input TCP port
int net_output_beast_port; // Beast output TCP port int net_output_beast_port; // Beast output TCP port
@ -375,9 +380,6 @@ struct { // Internal state
struct demod_stats stat_demod_phasecorrected; struct demod_stats stat_demod_phasecorrected;
unsigned int stat_http_requests; unsigned int stat_http_requests;
unsigned int stat_sbs_connections;
unsigned int stat_raw_connections;
unsigned int stat_beast_connections;
unsigned int stat_out_of_phase; unsigned int stat_out_of_phase;
unsigned int stat_DF_Len_Corrected; unsigned int stat_DF_Len_Corrected;
@ -473,11 +475,9 @@ struct stDF *interactiveFindDF (uint32_t addr);
// Functions exported from net_io.c // Functions exported from net_io.c
// //
void modesInitNet (void); void modesInitNet (void);
void modesReadFromClients (void);
void modesSendAllClients (int service, void *msg, int len);
void modesQueueOutput (struct modesMessage *mm); void modesQueueOutput (struct modesMessage *mm);
void modesReadFromClient(struct client *c, char *sep, int(*handler)(struct client *, char *)); void modesReadFromClient(struct client *c, char *sep, int(*handler)(struct client *, char *));
void modesNetCleanup (void); void modesNetPeriodicWork (void);
#ifdef __cplusplus #ifdef __cplusplus
} }

77
mode_s.c
View file

@ -1875,43 +1875,6 @@ void detectModeS(uint16_t *m, uint32_t mlen) {
use_correction = 0; use_correction = 0;
} }
} }
//Send any remaining partial raw buffers now
if (Modes.rawOutUsed || Modes.beastOutUsed)
{
Modes.net_output_raw_rate_count++;
if (Modes.net_output_raw_rate_count > Modes.net_output_raw_rate)
{
if (Modes.rawOutUsed) {
modesSendAllClients(Modes.ros, Modes.rawOut, Modes.rawOutUsed);
Modes.rawOutUsed = 0;
}
if (Modes.beastOutUsed) {
modesSendAllClients(Modes.bos, Modes.beastOut, Modes.beastOutUsed);
Modes.beastOutUsed = 0;
}
Modes.net_output_raw_rate_count = 0;
}
}
else if ( (Modes.net)
&& (Modes.net_heartbeat_rate)
&& ((++Modes.net_heartbeat_count) > Modes.net_heartbeat_rate) ) {
//
// We haven't received any Mode A/C/S messages for some time. To try and keep any 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.
//
// Fudge up a null message
memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = Modes.timestampBlk;
// Feed output clients
modesQueueOutput(&mm);
// Reset the heartbeat counter
Modes.net_heartbeat_count = 0;
}
} }
// 2.4MHz sampling rate version // 2.4MHz sampling rate version
@ -2404,43 +2367,6 @@ void detectModeS_oversample(uint16_t *m, uint32_t mlen) {
} }
} }
} }
//Send any remaining partial raw buffers now
if (Modes.rawOutUsed || Modes.beastOutUsed)
{
Modes.net_output_raw_rate_count++;
if (Modes.net_output_raw_rate_count > Modes.net_output_raw_rate)
{
if (Modes.rawOutUsed) {
modesSendAllClients(Modes.ros, Modes.rawOut, Modes.rawOutUsed);
Modes.rawOutUsed = 0;
}
if (Modes.beastOutUsed) {
modesSendAllClients(Modes.bos, Modes.beastOut, Modes.beastOutUsed);
Modes.beastOutUsed = 0;
}
Modes.net_output_raw_rate_count = 0;
}
}
else if ( (Modes.net)
&& (Modes.net_heartbeat_rate)
&& ((++Modes.net_heartbeat_count) > Modes.net_heartbeat_rate) ) {
//
// We haven't received any Mode A/C/S messages for some time. To try and keep any 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.
//
// Fudge up a null message
memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = Modes.timestampBlk;
// Feed output clients
modesQueueOutput(&mm);
// Reset the heartbeat counter
Modes.net_heartbeat_count = 0;
}
} }
// //
@ -2468,9 +2394,6 @@ void useModesMessage(struct modesMessage *mm) {
// Feed output clients // Feed output clients
if (Modes.net) {modesQueueOutput(mm);} if (Modes.net) {modesQueueOutput(mm);}
// Heartbeat not required whilst we're seeing real messages
Modes.net_heartbeat_count = 0;
} }
} }
// //

216
net_io.c
View file

@ -49,6 +49,7 @@
struct service { struct service {
char *descr; char *descr;
int *socket; int *socket;
struct net_writer *writer;
int port; int port;
int enabled; int enabled;
}; };
@ -59,12 +60,12 @@ void modesInitNet(void) {
int j; int j;
struct service svc[MODES_NET_SERVICES_NUM] = { struct service svc[MODES_NET_SERVICES_NUM] = {
{"Raw TCP output", &Modes.ros, Modes.net_output_raw_port, 1}, {"Raw TCP output", &Modes.raw_out.socket, &Modes.raw_out, Modes.net_output_raw_port, 1},
{"Raw TCP input", &Modes.ris, Modes.net_input_raw_port, 1}, {"Raw TCP input", &Modes.ris, NULL, Modes.net_input_raw_port, 1},
{"Beast TCP output", &Modes.bos, Modes.net_output_beast_port, 1}, {"Beast TCP output", &Modes.beast_out.socket, &Modes.beast_out, Modes.net_output_beast_port, 1},
{"Beast TCP input", &Modes.bis, Modes.net_input_beast_port, 1}, {"Beast TCP input", &Modes.bis, NULL, Modes.net_input_beast_port, 1},
{"HTTP server", &Modes.https, Modes.net_http_port, 1}, {"HTTP server", &Modes.https, NULL, Modes.net_http_port, 1},
{"Basestation TCP output", &Modes.sbsos, Modes.net_output_sbs_port, 1} {"Basestation TCP output", &Modes.sbs_out.socket, &Modes.sbs_out, Modes.net_output_sbs_port, 1}
}; };
memcpy(&services, &svc, sizeof(svc));//services = svc; memcpy(&services, &svc, sizeof(svc));//services = svc;
@ -93,6 +94,18 @@ void modesInitNet(void) {
} }
anetNonBlock(Modes.aneterr, s); anetNonBlock(Modes.aneterr, s);
*services[j].socket = 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 = time(NULL);
}
} else { } else {
if (Modes.debug & MODES_DEBUG_NET) printf("%s port is disabled\n", services[j].descr); if (Modes.debug & MODES_DEBUG_NET) printf("%s port is disabled\n", services[j].descr);
} }
@ -127,9 +140,11 @@ struct client * modesAcceptClients(void) {
Modes.clients = c; Modes.clients = c;
anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size)); anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size));
if (*services[j].socket == Modes.sbsos) Modes.stat_sbs_connections++; if (services[j].writer) {
if (*services[j].socket == Modes.ros) Modes.stat_raw_connections++; if (++ services[j].writer->connections == 1) {
if (*services[j].socket == Modes.bos) Modes.stat_beast_connections++; services[j].writer->lastWrite = time(NULL); // suppress heartbeat initially
}
}
j--; // Try again with the same listening port j--; // Try again with the same listening port
@ -145,18 +160,21 @@ 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 modesFreeClient(struct client *c) { void modesFreeClient(struct client *c) {
int j;
// 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
// client (unpredictably: reading from client A may cause client B to // client (unpredictably: reading from client A may cause client B to
// be freed) // be freed)
close(c->fd); close(c->fd);
if (c->service == Modes.sbsos) {
if (Modes.stat_sbs_connections) Modes.stat_sbs_connections--; for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
} else if (c->service == Modes.ros) { if (c->service == *services[j].socket) {
if (Modes.stat_raw_connections) Modes.stat_raw_connections--; if (services[j].writer)
} else if (c->service == Modes.bos) { services[j].writer->connections--;
if (Modes.stat_beast_connections) Modes.stat_beast_connections--; break;
}
} }
if (Modes.debug & MODES_DEBUG_NET) if (Modes.debug & MODES_DEBUG_NET)
@ -169,36 +187,72 @@ void modesFreeClient(struct client *c) {
// //
//========================================================================= //=========================================================================
// //
// Send the specified message to all clients listening for a given service // Send the write buffer for the specified writer to all connected clients
// //
void modesSendAllClients(int service, void *msg, int len) { 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 == service) { if (c->service == writer->socket) {
#ifndef _WIN32 #ifndef _WIN32
int nwritten = write(c->fd, msg, len); int nwritten = write(c->fd, writer->data, writer->dataUsed);
#else #else
int nwritten = send(c->fd, msg, len, 0 ); int nwritten = send(c->fd, writer->data, writer->dataUsed, 0 );
#endif #endif
if (nwritten != len) { if (nwritten != writer->dataUsed) {
modesFreeClient(c); modesFreeClient(c);
} }
} }
} }
writer->dataUsed = 0;
writer->lastWrite = time(NULL);
} }
// Prepare to write up to 'len' bytes to the given net_writer.
// Returns a pointer to write to, or NULL to skip this write.
static void *prepareWrite(struct net_writer *writer, int len) {
if (!writer ||
!writer->connections ||
!writer->data)
return NULL;
if (len >= MODES_OUT_BUF_SIZE)
return NULL;
if (writer->dataUsed + len >= MODES_OUT_BUF_SIZE) {
// Flush now to free some space
flushWrites(writer);
}
return writer->data + writer->dataUsed;
}
// Complete a write previously begun by prepareWrite.
// endptr should point one byte past the last byte written
// to the buffer returned from prepareWrite.
static void completeWrite(struct net_writer *writer, void *endptr) {
writer->dataUsed = endptr - writer->data;
if (writer->dataUsed >= Modes.net_output_flush_size) {
flushWrites(writer);
}
}
// //
//========================================================================= //=========================================================================
// //
// 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) { void modesSendBeastOutput(struct modesMessage *mm) {
char *p = &Modes.beastOut[Modes.beastOutUsed];
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.beast_out, 2 + 2 * (7 + msgLen));
char * pTimeStamp; char * pTimeStamp;
char ch; char ch;
int j; int j;
int iOutLen = msgLen + 9; // Escape, msgtype, timestamp, sigLevel, msg
if (!p)
return;
*p++ = 0x1a; *p++ = 0x1a;
if (msgLen == MODES_SHORT_MSG_BYTES) if (msgLen == MODES_SHORT_MSG_BYTES)
@ -213,36 +267,34 @@ void modesSendBeastOutput(struct modesMessage *mm) {
pTimeStamp = (char *) &mm->timestampMsg; pTimeStamp = (char *) &mm->timestampMsg;
for (j = 5; j >= 0; j--) { for (j = 5; j >= 0; j--) {
*p++ = (ch = pTimeStamp[j]); *p++ = (ch = pTimeStamp[j]);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
} }
*p++ = (ch = mm->signalLevel); *p++ = (ch = mm->signalLevel);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
for (j = 0; j < msgLen; j++) { for (j = 0; j < msgLen; j++) {
*p++ = (ch = mm->msg[j]); *p++ = (ch = mm->msg[j]);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
} }
Modes.beastOutUsed += iOutLen; completeWrite(&Modes.beast_out, p);
if (Modes.beastOutUsed >= Modes.net_output_raw_size)
{
modesSendAllClients(Modes.bos, Modes.beastOut, Modes.beastOutUsed);
Modes.beastOutUsed = 0;
Modes.net_output_raw_rate_count = 0;
}
} }
// //
//========================================================================= //=========================================================================
// //
// Write raw output to TCP clients // Write raw output to TCP clients
// //
void modesSendRawOutput(struct modesMessage *mm) { void modesSendRawOutput(struct modesMessage *mm) {
char *p = &Modes.rawOut[Modes.rawOutUsed];
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.raw_out, msgLen*2 + 15);
int j; int j;
unsigned char * pTimeStamp; unsigned char * pTimeStamp;
if (!p)
return;
if (Modes.mlat && mm->timestampMsg) { if (Modes.mlat && mm->timestampMsg) {
*p++ = '@'; *p++ = '@';
pTimeStamp = (unsigned char *) &mm->timestampMsg; pTimeStamp = (unsigned char *) &mm->timestampMsg;
@ -250,7 +302,6 @@ void modesSendRawOutput(struct modesMessage *mm) {
sprintf(p, "%02X", pTimeStamp[j]); sprintf(p, "%02X", pTimeStamp[j]);
p += 2; p += 2;
} }
Modes.rawOutUsed += 12; // additional 12 characters for timestamp
} else } else
*p++ = '*'; *p++ = '*';
@ -262,13 +313,7 @@ void modesSendRawOutput(struct modesMessage *mm) {
*p++ = ';'; *p++ = ';';
*p++ = '\n'; *p++ = '\n';
Modes.rawOutUsed += ((msgLen*2) + 3); completeWrite(&Modes.raw_out, p);
if (Modes.rawOutUsed >= Modes.net_output_raw_size)
{
modesSendAllClients(Modes.ros, Modes.rawOut, Modes.rawOutUsed);
Modes.rawOutUsed = 0;
Modes.net_output_raw_rate_count = 0;
}
} }
// //
//========================================================================= //=========================================================================
@ -277,17 +322,27 @@ void modesSendRawOutput(struct modesMessage *mm) {
// 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) { void modesSendSBSOutput(struct modesMessage *mm) {
char msg[256], *p = msg; char *p = prepareWrite(&Modes.sbs_out, 200);
uint32_t offset; uint32_t offset;
struct timeb epocTime_receive, epocTime_now; struct timeb epocTime_receive, epocTime_now;
struct tm stTime_receive, stTime_now; struct tm stTime_receive, stTime_now;
int msgType; int msgType;
if (!p)
return;
// //
// SBS BS style output checked against the following reference // SBS BS style output checked against the following reference
// http://www.homepages.mcb.net/bones/SBS/Article/Barebones42_Socket_Data.htm - seems comprehensive // http://www.homepages.mcb.net/bones/SBS/Article/Barebones42_Socket_Data.htm - seems comprehensive
// //
if (mm->msgtype == -1) {
// heartbeat
p += sprintf(p, "\r\n");
completeWrite(&Modes.sbs_out, p);
return;
}
// Decide on the basic SBS Message Type // Decide on the basic SBS Message Type
if ((mm->msgtype == 4) || (mm->msgtype == 20)) { if ((mm->msgtype == 4) || (mm->msgtype == 20)) {
msgType = 5; msgType = 5;
@ -434,15 +489,16 @@ void modesSendSBSOutput(struct modesMessage *mm) {
} }
p += sprintf(p, "\r\n"); p += sprintf(p, "\r\n");
modesSendAllClients(Modes.sbsos, msg, p-msg);
completeWrite(&Modes.sbs_out, p);
} }
// //
//========================================================================= //=========================================================================
// //
void modesQueueOutput(struct modesMessage *mm) { void modesQueueOutput(struct modesMessage *mm) {
if (Modes.stat_sbs_connections) {modesSendSBSOutput(mm);} modesSendSBSOutput(mm);
if (Modes.stat_beast_connections) {modesSendBeastOutput(mm);} modesSendBeastOutput(mm);
if (Modes.stat_raw_connections) {modesSendRawOutput(mm);} modesSendRawOutput(mm);
} }
// //
//========================================================================= //=========================================================================
@ -928,17 +984,21 @@ void modesReadFromClient(struct client *c, char *sep,
} }
} }
} }
//
//=========================================================================
//
// Read data from clients. This function actually delegates a lower-level
// function that depends on the kind of service (raw, http, ...).
//
void modesReadFromClients(void) {
struct client *c = modesAcceptClients(); //
// Perform periodic network work
//
void modesNetPeriodicWork(void) {
struct client *c, **prev;
time_t now = time(NULL);
int j;
int need_heartbeat = 0, need_flush = 0;
for (; c; c = c->next) { // Accept new connetions
modesAcceptClients();
// Read from clients
for (c = Modes.clients; c; c = c->next) {
if (c->service == Modes.ris) { if (c->service == Modes.ris) {
modesReadFromClient(c,"\n",decodeHexMessage); modesReadFromClient(c,"\n",decodeHexMessage);
} else if (c->service == Modes.bis) { } else if (c->service == Modes.bis) {
@ -947,14 +1007,52 @@ void modesReadFromClients(void) {
modesReadFromClient(c,"\r\n\r\n",handleHTTPRequest); modesReadFromClient(c,"\r\n\r\n",handleHTTPRequest);
} }
} }
// If we have generated no messages for a while, generate
// a dummy heartbeat message.
if (Modes.net_heartbeat_interval) {
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
if (services[j].writer &&
services[j].writer->connections &&
(services[j].writer->lastWrite + Modes.net_heartbeat_interval) <= now) {
need_flush = 1;
if (services[j].writer->dataUsed == 0) {
need_heartbeat = 1;
break;
}
}
}
} }
if (need_heartbeat) {
// //
// Perform cleanup of any recently-closed clients. // 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
// link which will cause an un-recoverable link error if/when a real frame arrives.
// //
void modesNetCleanup(void) { // Fudge up a null message
struct client *c, **prev; struct modesMessage mm;
memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = 0;
mm.msgtype = -1;
// Feed output clients
modesQueueOutput(&mm);
}
// If we have data that has been waiting to be written for a while,
// write it now.
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
if (services[j].writer &&
services[j].writer->dataUsed &&
(need_flush || (services[j].writer->lastWrite + Modes.net_output_flush_interval) <= now)) {
flushWrites(services[j].writer);
}
}
// Unlink and free closed clients
for (prev = &Modes.clients, c = *prev; c; c = *prev) { for (prev = &Modes.clients, c = *prev; c; c = *prev) {
if (c->fd == -1) { if (c->fd == -1) {
// Recently closed, prune from list // Recently closed, prune from list