popufare/busunit/DIUv2/diu_main.c

/*
 * Copyright (c) 2019 Clementine Computing LLC.
 *
 * This file is part of PopuFare.
 *
 * PopuFare is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * PopuFare 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 Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with PopuFare.  If not, see <https://www.gnu.org/licenses/>.
 *
 */

#include <sys/time.h>
#include <stdio.h>
#include <stdlib.h>
#include <poll.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <signal.h>
#include <time.h>

#include <openssl/md5.h>

#include <ctype.h>
#include <string.h>

#include <sys/types.h>
#include <ifaddrs.h>

#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>
#include <ifaddrs.h>
#include <linux/if_link.h>


#include "../common/common_defs.h"
#include "../commhub/commhub.h"
#include "../commhub/client_utils.h"

#include "driver.h"
#include "touchscreen.h"

#include "mongoose.h"

#define _SLEN LINE_BUFFER_SIZE

static const char *s_http_port = "60535";
static struct mg_serve_http_opts s_http_server_opts;

struct mg_mgr g_mgr;

int diu_fd = -1;
int hub_fd = -1;

time_t last_hub_try = 0;
time_t last_diu_clock = 0;

gps_status my_gps_stat={0};

time_t mkgmtime(struct tm *tm);

int   token_diag_serial = 0;
char   token_diag_string[LINE_BUFFER_SIZE] = {0};


// return total bytes in dst
//
static size_t _strnncat(char *dst, char *src, size_t n) {
  size_t dst_n, src_n, N, i;

  dst_n = strlen(dst);
  src_n = strlen(src);
  N = n-1;
  if (dst_n < N) { N -= dst_n; }

  for (i=0; (i<N) && (i<src_n); i++) {
    dst[dst_n + i] = src[i];
  }

  dst[dst_n+i] = '\0';

  return dst_n + i;
}

time_t paddle_req_timeout = 0;
set_paddle_req paddle_req = {0};

driver_status my_driver_status={0};
int update_driver_status = 0;

int md5_of_file(char *filename, void *result) {
  FILE *f = NULL;
  char chunk[LINE_BUFFER_SIZE]={0};
  int retval = 0;
  MD5_CTX ctx;

  if (!result) { return -1; }

  if (!filename) { return -1; }

  if(!MD5_Init(&ctx)) {
    return -1;
  }

  f = fopen(filename, "rb");

  if(!f) {
    return -1;
  }

  while(1) {
    retval = fread(chunk, 1, LINE_BUFFER_SIZE, f);

    if(retval <= 0) {
      break;
    }

    else {
      MD5_Update(&ctx, chunk, retval);
    }
  }

  fclose(f);

  MD5_Final((unsigned char *)result, &ctx);

  return 0;
}

//--------------------

static int is_websocket(const struct mg_connection *nc) {
  return nc->flags & MG_F_IS_WEBSOCKET;
}

void ws_send(struct mg_mgr *mgr, char *msg) {
  struct mg_connection *conn;
  for (conn = mg_next(mgr, NULL); conn; conn = mg_next(mgr, conn)) {
    if (!is_websocket(conn)) { continue; }
    mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT, msg, strlen(msg));
  }
}

//--------------------

void beep(int fd, int hz, int milis) {
  char buffer[64];
  int n;

  n = sprintf(buffer, "/B:%X,%X\r", hz, milis);

  write(fd,buffer,n);
}

void set_backlight(int fd, int on) {
  char buffer[64];
  int n;

  n = sprintf(buffer, "/D:%c\r", on?'0':'1');

  write(fd,buffer,n);
}

void set_diu_clock(int fd, time_t tt) {
  struct tm t;
  char buffer[16];
  int n;

  localtime_r(&tt, &t);

  n = sprintf(buffer, "/C:%02d%02d\r", t.tm_hour, t.tm_min);

  write(fd, buffer, n);
}

//--------------------

void clear_diu_messages() {
}


//--------------------


//    This function takes a GPS timestamp (in gross GPS float format + gross integer date) and
//makes it into a sane UTC timestamp.  If we are more than MAX_GPS_CLOCK_DRIFT seconds off
//from GPS time, it sets the system clock to GPS time.
//
int handle_gps_time(float gtime, int date) {
  int day,month,year;
  int hour,min,sec,frac;
  int n = 0;
  time_t utc,now;

  char buffer[32] = {0};

  struct message_record outgoing_msg;

  now = time(NULL);

  day = month = year = 0;
  hour = min = sec = frac = 0;

  //    Just to be *ahem* clear, as per NMEA standard the date is encoded DDMMYY, and the time of day
  // is encoded HHMMSS[.frac] where there is an optional fractional second field denoted by a decimal point.
  // we must be able to decode the fractional seconds field but we discard the information since it is not
  // reliable enough to be worth the bother.
  //    Whoever thought that NMEA GPS should encode the time in this format ought to be shot...


  // Start out with zero parsed fields
  //
  n = 0;

  //    Construct and then re-parse the time from its icky format to discrete values (this should result
  // in at least three (possibly four) fields.
  //
  sprintf(buffer,"%010.3f", gtime);
  n += sscanf(buffer,"%02d%02d%02d.%d", &hour, &min, &sec, &frac);

  // Construct and then re-parse the date from its icky format to discrete values.  This should result in three fields.
  //
  sprintf(buffer,"%06d", date);
  n += sscanf(buffer,"%02d%02d%02d", &day, &month, &year);

  // If we scanned at all required fields
  //
  if(n >= 6) {
    struct tm gpstm = {0};

    // GPS date only uses two digits for year, so we must assume the century
    //
    year += GPS_DATE_CENTURY;

    // tm.tm_year is based on the year 1900
    //
    gpstm.tm_year = year - 1900;

    // January = month 0 in struct tm.tm_mon whereas January = month 1 in an NMEA GPS date.
    //
    gpstm.tm_mon = month - 1;

    gpstm.tm_mday = day;
    gpstm.tm_hour = hour;
    gpstm.tm_min = min;
    gpstm.tm_sec = sec;

    // Go and turn the struct tm we've just populated into an utc time stamp (seconds since epoch)
    //
    utc = mkgmtime(&gpstm);

    // Most importantly... Remember what the self-reported GPS time stamp is.
    //
    my_gps_stat.gpstime = utc;

    //printf("%02d-%02d-%04d %02d:%02d:%02d\n",day,month,year,hour,min,sec);
    //printf("CALCULATED: %d\nSYSTEM    : %d\n\n",(int)utc,(int)time(NULL));

    // if we have more than MAX_GPS_CLOCK_DRIFT seconds of clock drift
    //
    if(abs(utc - now) > MAX_GPS_CLOCK_DRIFT) {
      struct timeval ts = {0};

      // Set the timeval struct to the calculated utc timestamp from the GPS unit.
      //
      ts.tv_sec = utc;
      ts.tv_usec = 0;

      // Set the system clock from GPS time using said timeval struct.
      //
      settimeofday(&ts, NULL);

      // If we have a valid connection to the IPC hub
      //
      if( hub_fd >= 0) {

        // Stick a message into the diagnostic log to record the fact that we've set the system clock from GPS
        //
        format_log_message(&outgoing_msg, LOGLEVEL_DEBUG, "Set syatem clock from previous value (%d) to GPS time (%d).", (int)now, (int)utc );
        send_message(hub_fd, &outgoing_msg);

      }

    }

  }

  return 0;
}


#ifdef CLEAR_GPS_ON_STALE
    static inline void clear_stale_gps_data()
    {
        my_gps_stat.lat = my_gps_stat.lon = my_gps_stat.heading = my_gps_stat.velocity = 0;
        my_gps_stat.num_sats = 0;
    }
#else
    static inline void clear_stale_gps_data()
    {
    }
#endif

static int handle_stale_gps_condition() {
  //     This return code will be > 0 if this function generates a status change that
  // will then need to be communicated to other modules in the system via a message to
  // MAILBOX_GPS_STATUS.
  //
  int return_code = 0;

  //     This will hold the gps_good flag as it stood at the beginning of this subroutine
  // previous to any adjustments we make.
  //
  int previous_good_flag = my_gps_stat.gps_good;

  int stale_time = 0;

  time_t now = time(NULL);

  stale_time = (now - my_gps_stat.stamp);

  // If we have entered this function with the impression that we have a valid GPS fix...
  //
  if(previous_good_flag > 0) {
    // If it has been at least GPS_STALE_THRESHOLD seconds since the last fix
    //
    if( stale_time >= GPS_STALE_THRESHOLD ) {
      clear_stale_gps_data();
      my_gps_stat.gps_good = 0;
      return_code |= 1;
    }

  }

  //     If we have determined that we need to declare the GPS data stale and invalid and
  // we have a valid connection to the IPC hub we should use that IPC hub connection to
  // add a note to the diagnostic log indicating that we've declared the GPS data stale.
  //
  if( return_code && (hub_fd >= 0) ) {
    struct message_record outgoing_msg;

    format_log_message(&outgoing_msg, LOGLEVEL_DEBUG, "GPS fix has been stale for %d seconds, setting GPS = NO.", stale_time);
    send_message(hub_fd, &outgoing_msg);
  }

  return return_code;
}

int update_gps(char *in) {
  //     This will hold the number of matched variables populated by sscanf().
  //
  int num = 0;

  //     This will allow us to know if we have made a transition from an invalid
  // GPS fix to a valid one so that we can log this information.
  //
  int previous_good_flag = my_gps_stat.gps_good;

  //     This return code will be > 0 if this function generates a status change that
  // will then need to be communicated to other modules in the system via a message to
  // MAILBOX_GPS_STATUS.
  //
  int return_code = 0;


  if(!strncmp(in,"$GPRMC",6)) {
    float f1=0;
    char f2=0;
    float f3=0;
    char f4=0;
    float f5=0;
    char f6=0;
    float f7=0;
    float f8=0;
    int f9=0;
    float f10=0;
    char f11=0;

    num = sscanf(in,"$GPRMC,%f,%c,%f,%c,%f,%c,%f,%f,%d,%f,%c",&f1,&f2,&f3,&f4,&f5,&f6,&f7,&f8,&f9,&f10,&f11);

    // If we have a full GPRMC sentence we can consider setting the time
    //
    if(num == 11) {

      //     Require at least MIN_SATS_FOR_TIME satellites to accept a new system clock value from the GPS unit.
      // This is to keep a crummy GPS fix from generating a bogus or unstable system time.
      //
      if(my_gps_stat.num_sats >= MIN_SATS_FOR_TIME) {
        //     Pass the time field (f1) and the date field (f9) in to the routine that sets the system clock if needed.
        // (this routine also stores the utc timestamp derived from the GPS date and time fields so it can be passed to
        // other modules that may have a need for this information).
        //handle_gps_time(f1,f9);
      }
    }

    if(num > 0) {

      // update snapshot with latitude
      //
      my_gps_stat.lat = f3 * ((f4 == 'N')?(1):(-1));

      // longitude
      //
      my_gps_stat.lon = f5 * ((f6 == 'E')?(1):(-1));

      // meters per second (converted from knots)
      //
      my_gps_stat.velocity = f7 / 1.94384449f;

      // course
      //
      my_gps_stat.heading = f8;

      // update snapshot's staledate
      //
      my_gps_stat.stamp = time(NULL);

      return_code |= 1;
    }
  }

  else if(!strncmp(in,"$GPGGA",6)) {
    int f1=0;
    float f2=0;
    char f3=0;
    float f4=0;
    char f5=0;
    int f6=0;
    int f7=0;
    float f8=0;
    float f9=0;
    char f10=0;
    float f11=0;
    char f12=0;

    num=sscanf(in,"$GPGGA,%d,%f,%c,%f,%c,%d,%d,%f,%f,%c,%f,%c",&f1,&f2,&f3,&f4,&f5,&f6,&f7,&f8,&f9,&f10,&f11,&f12);

    if(num == 12) {

      // require 3 satellites minimum
      //
      if ( f7 >= 3 ) {

        // store GPS valid flag in snapshot
        //
        my_gps_stat.gps_good = f6;

        // store number of satellites in view in snapshot
        //
        my_gps_stat.num_sats = f7;

        //     Do NOT store the timestamp since we only want to remember timestamps of position
        // fixes and the GPGGA message is all metadata (number of satellites, fix quality, etc...).
        // It is worth noting that GPGGA does report altitude, but that is not a piece of information
        // we track because the road is where the road is, and if a bus becomes airborn we have much
        // bigger problems than figuring out how far off the ground it is...
        //

        return_code |= 1;
      }
    }

  }

  return_code |= handle_stale_gps_condition();

  //     If we have a connection to the IPC hub and we had previously not had a valid
  // GPS fix but we now do, make a note of it in the diagnostic log.
  //
  if( (previous_good_flag == 0) && (my_gps_stat.gps_good != 0) && (hub_fd >= 0) ) {
    struct message_record outgoing_msg;

    format_log_message(&outgoing_msg, LOGLEVEL_DEBUG, "GPS fix is now valid with %d satellites. Setting GPS = YES.", my_gps_stat.num_sats);
    send_message(hub_fd, &outgoing_msg);
  }

  return return_code;
}

void maintain_ipc_hub_connect(char *progname) {
  struct message_record outgoing_msg;

  // if we have no connection to the IPC hub
  //
  if(hub_fd < 0) {

    // if we haven't tried the hub in a few seconds
    //
    if( (time(NULL) - last_hub_try) > HUB_RETRY_TIME ) {

      // retry it
      //
      last_hub_try = time(NULL);

      // try and get one
      //
      hub_fd = connect_to_message_server(progname);

      if(hub_fd >= 0) {
        // Subscribe to the default status messages
        //
        subscribe_to_default_messages(hub_fd);

        // Subscribe to our specific message
        //
        prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_DRIVER_NOTIFY, strlen(MAILBOX_DRIVER_NOTIFY));
        send_message(hub_fd,&outgoing_msg);

        prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_PADDLE_ACK, strlen(MAILBOX_PADDLE_ACK));
        send_message(hub_fd,&outgoing_msg);

        prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_VAULT_DROP, strlen(MAILBOX_VAULT_DROP));
        send_message(hub_fd,&outgoing_msg);

        prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_TOKEN_MAG, strlen(MAILBOX_TOKEN_MAG));
        send_message(hub_fd,&outgoing_msg);

        prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_TOKEN_RFID, strlen(MAILBOX_TOKEN_RFID));
        send_message(hub_fd,&outgoing_msg);

        // Ask for a status update
        //
        prepare_message(&outgoing_msg, MAILBOX_STATUS_REQUEST, "", 0);
        send_message(hub_fd,&outgoing_msg);
      }
      else {
        fprintf(stderr, "Cannot connect to IPC hub!\n");
      }
    }
  }
}

message_callback_return handle_status_request(struct message_record *msg, void *param) {
  struct message_record outgoing_msg;

  if(hub_fd >= 0) {
    prepare_message(&outgoing_msg, MAILBOX_DRIVER_STATUS, &my_driver_status, sizeof(my_driver_status));
    send_message(hub_fd, &outgoing_msg);
    prepare_message(&outgoing_msg, MAILBOX_GPS_STATUS, &my_gps_stat, sizeof(my_gps_stat));
    send_message(hub_fd, &outgoing_msg);
  }

  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_vault_drop(struct message_record *msg, void *param) {
  if(diu_fd >= 0) { write(diu_fd, "/V:\r", 4); }
  return MESSAGE_HANDLED_CONT;
}

char dup_notify_str[MAX_PAYLOAD_LENGTH] = {0};
int dup_notify_count = 0;
long long dup_notify_usec = 0;


message_callback_return handle_driver_notify(struct message_record *msg, void *param) {
  int is_dup;
  char _text[MAX_PAYLOAD_LENGTH] = {0};
  long long dup_usec_delta = 0;


  if(strncmp((const char *)(msg->payload), dup_notify_str, MAX_PAYLOAD_LENGTH)) {
    dup_notify_count = 1;
    strncpy(dup_notify_str, (const char *)(msg->payload), MAX_PAYLOAD_LENGTH - 1);
    dup_notify_str[MAX_PAYLOAD_LENGTH - 1] = '\0';
    is_dup = 0;
    dup_notify_usec = 0;
  }
  else {
    dup_notify_count++;
    is_dup = 1;
    dup_usec_delta = get_usec_time() - dup_notify_usec;
    dup_notify_usec = get_usec_time();
  }

  switch(msg->payload[0]) {
    case LOGLEVEL_EVENT:
      if(!is_dup || (dup_usec_delta >= DUP_USEC_BEEP_THRESHOLD)) {
        DIU_ACK_BEEP(diu_fd);
      }
      break;

      case LOGLEVEL_REJECT:
        if(!is_dup || (dup_usec_delta >= DUP_USEC_BEEP_THRESHOLD)) {
          DIU_ERROR_BEEP(diu_fd);
        }
        break;

      case LOGLEVEL_ACCEPT:
        if(!is_dup || (dup_usec_delta >= DUP_USEC_BEEP_THRESHOLD)) {
          DIU_ACK_BEEP(diu_fd);
        }
        break;

      case LOGLEVEL_ERROR:
        if(!is_dup || (dup_usec_delta >= DUP_USEC_BEEP_THRESHOLD)) {
          DIU_CRITICAL_BEEP(diu_fd);
        }
        break;

      default: break;
  }

  if(is_dup) {
    snprintf(_text, MAX_PAYLOAD_LENGTH, "driver_notify replace %s %s %d x %s", "white", "black", dup_notify_count, &msg->payload[1]);
    ws_send(&g_mgr, _text);
  }
  else {
    snprintf(_text, MAX_PAYLOAD_LENGTH, "driver_notify ok %s %s %d x %s", "white", "black", dup_notify_count, &msg->payload[1]);
    ws_send(&g_mgr, _text);
  }

  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_paddle_ack(struct message_record *msg, void *param) {
  set_paddle_req *pr = (set_paddle_req *)msg->payload;
  paddle_req.result = pr->result;
  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_token_diag(struct message_record *msg, void *param) {
  strncpy(token_diag_string, (const char *)(msg->payload), sizeof(token_diag_string));
  token_diag_string[sizeof(token_diag_string) - 1] = '\0';
  token_diag_serial++;
  return MESSAGE_HANDLED_CONT;
}

//----------------------------------
time_t diu_error_burst = 0;
int    diu_error_counter = 0;

static inline int can_report_diu_error() {
  time_t now = time(NULL);

  // If our last potential burst lockout has expired...
  //
  if( (now - diu_error_burst) >= DIU_ERROR_RATE_LIMIT) {

    // reset our burst counter
    //
    diu_error_counter = 0;

    // and start the "potential burst" time at this message
    //
    diu_error_burst = now;
  }

  // if we haven't hit our burst limit...
  //
  if(diu_error_counter < DIU_ERROR_BURST_LIMIT) {

    // count this message against our burst limit
    //
    diu_error_counter++;

    // and allow it to pass
    //
    return 1;
  }

  // if we have hit our limit
  //
  else {

    // ignore this message
    //
    return 0;
  }
}

//----------------------------------

//-------- web server functions ----

/*
static void send_ws_heartbeat(struct mg_connection *nc) {
  int n;
  char buf[_SLEN];
  struct mg_connection *c;

  for (c=mg_next(nc->mgr, NULL); c ; c = mg_next(nc->mgr, c)) {
    if (c==nc) { continue; }
    if (!is_websocket(c)) { continue; }
    printf("...\n");

    snprintf(buf, _SLEN, "hello");
    n = strlen(buf);
    mg_send_websocket_frame(c, WEBSOCKET_OP_TEXT, buf, n);
  }

}
*/

static void process_ws_message(struct websocket_message *ws_msg) {
  int i;
  char *data;

  data = (char *)(ws_msg->data);
  printf("#ws, got(%i)\n:", (int)(ws_msg->size));
  for (i=0; i<ws_msg->size; i++) {
    printf("%c", data[i]);
  }
  printf("\n");
}

#define MAX_PKGS  (16)

//
// ui wants status information
//
static void ui_handle_status_input(struct mg_connection *nc, struct http_message *hm) {
  int i, npkgs, npkgnet=0;
  char buf[4*_SLEN],
       pkgnetline[_SLEN],
       _str[_SLEN];
  char date_str[32];
  time_t t;
  struct tm tm, pkgtime;
  FILE *fp;
  package_signature pkgs[MAX_PKGS];

  t = time(NULL);
  localtime_r(&t, &tm);
  strftime(date_str, 32, "%Y-%m-%d", &tm);

  pkgnetline[0] = '\0';
  npkgs = find_packages(pkgs,MAX_PKGS);
  for (i=0; i<npkgs; i++) {
    localtime_r(&pkgs[i].installed, &pkgtime);
    snprintf(_str, _SLEN, "pkg%i=%s %s %02d/%02d/%02d %02d:%02d:%02d",
                          i,
                          pkgs[i].pkgname, pkgs[i].pkgver,
                          pkgtime.tm_mon + 1, pkgtime.tm_mday, pkgtime.tm_year + 1900,
                          pkgtime.tm_hour, pkgtime.tm_min, pkgtime.tm_sec);

		_strnncat(pkgnetline, "|", _SLEN);
    _strnncat(pkgnetline, _str, _SLEN);
    npkgnet++;
  }

  fp = fopen("/tmp/net_ids", "rb");
  if (fp) {
    i=0;
    while (fgets(_str, _SLEN, fp)) {
      strip_crlf(_str);
      _strnncat(pkgnetline, "|", _SLEN);
      _strnncat(pkgnetline, _str, _SLEN);
      npkgnet++;
    }
    fclose(fp);
  }


  snprintf(buf, 4*_SLEN, "ok msg=status\n"
      "equipno=%d|"

      "route=%d|"
      "trip=%d|"
      "stop=%d|"

      "gps=%i|"
      "gpslat=%f|"
      "gpslon=%f|"
      "gpsv=%f|"
      "gpsheading=%f|"

      "tunnel=%i|"
      "gprs=%i|"

      "date=%s|"
      "nmsg=%d|"
      "last_token=%d"

      "%s",

      get_equip_num(),

      stop_stat.route,
      stop_stat.trip,
      stop_stat.stop,

      gps_stat.gps_good,
      gps_stat.lat,
      gps_stat.lon,
      gps_stat.velocity,
      gps_stat.heading,

      tunnel_is_up(),
      gprs_is_up(),

      date_str,
      bill_stat.unsynced_messages,
      token_diag_serial % 100,

      pkgnetline);

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(buf), buf);
}



// driver logout
//
static void ui_handle_logout_input(struct mg_connection *nc, struct http_message *hm) {
  char msg[][_SLEN] = {
    "fail error",
    "ok .",
  };

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg[1]), msg[1]);

}

// (manual) next stop pressed
//
static void ui_handle_nextstop_input(struct mg_connection *nc, struct http_message *hm) {
  char msg[][_SLEN] = {
    "fail error",
    "ok .",
  };

  action_nextstop();
  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg[1]), msg[1]);

}

// (manual) previous stop pressed
//
static void ui_handle_prevstop_input(struct mg_connection *nc, struct http_message *hm) {
  char msg[][_SLEN] = {
    "fail error",
    "ok .",
  };

  action_prevstop();
  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg[1]), msg[1]);

}

// generic fare input
//
static void ui_handle_fare_input(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_rule[_SLEN], s_param[_SLEN];
  char msg_ok[] = "ok .";

  driver_rulecall dr = {{0}};
  struct message_record outgoing;

  ret = mg_get_http_var(&(hm->body), "rule", s_rule, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  ret = mg_get_http_var(&(hm->body), "param", s_param, _SLEN);
  if (ret<=0) { s_param[0] = '\0'; }

  if (hub_fd >= 0) {
    strncpy(dr.rulename, s_rule, DRIVER_RULECALL_LEN - 1);
    strncpy(dr.ruleparam, s_param, DRIVER_RULECALL_LEN - 1);

    prepare_message(&outgoing, MAILBOX_RULE_CALL, &dr, sizeof(dr));
    send_message(hub_fd, &outgoing);
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_ok), msg_ok);
}



// driver paddle (route) input
//
static void ui_handle_paddle_input(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_paddle[_SLEN];

  //char msg_fail[] = "fail .";
  char msg_ok[] = "ok .";

  int ipaddle;

  ret = mg_get_http_var(&(hm->body), "paddle", s_paddle, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  //DEBUG
  printf("#got paddle %s\n", s_paddle);

  ipaddle = atoi(s_paddle);
  make_paddle_request(ipaddle);

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_ok), msg_ok);
}

// driver login
//
static void ui_handle_driver_login(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_driver[_SLEN], s_pin[_SLEN];
  int idrv=-1;

  char msg_fail[] = "fail driver_login";
  char msg_success[] = "ok driver";

  ret = mg_get_http_var(&(hm->body), "driver", s_driver, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  ret = mg_get_http_var(&(hm->body), "pin", s_pin, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  idrv = atoi(s_driver);
  ret = driver_login(idrv, s_pin);

  if (ret!=0) {
    mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
        (unsigned long)strlen(msg_fail), msg_fail);
    return;
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_success), msg_success);
}

// screen dim event
//
static void ui_handle_dim(struct mg_connection *nc, struct http_message *hm) {
  int ret;

  char msg_fail[] = "fail dim";
  char msg_success[] = "ok dim";

  // assuming we're running as root for now
  //
  ret = system("su - pi -c \"/usr/bin/xset -display :0 dpms force off\" &");

  if (ret!=0) {
    mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
        (unsigned long)strlen(msg_fail), msg_fail);
    return;
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_success), msg_success);
}

// set volume
//
static void ui_handle_volume(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_volume[_SLEN], _cmd[1024];
  float f_vol = 100.0;

  char msg_fail[] = "fail volume";
  char msg_success[] = "ok volume";

  ret = mg_get_http_var(&(hm->body), "volume", s_volume, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  // assuming we're running as root for now
  //
  f_vol = atof(s_volume);
  snprintf(_cmd, 1023, "/home/bus/bin/volumeset %04f &", f_vol);
  ret = system(_cmd);

  if (ret!=0) {
    mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
        (unsigned long)strlen(msg_fail), msg_fail);
    return;
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_success), msg_success);
}

// bring network interface up/down
//
static void ui_handle_interfaceupdown(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_interface[_SLEN], s_updown[_SLEN];

  char msg_fail[] = "fail interfaceupdown";
  char msg_success[] = "ok interfaceupdown";

  int _iface_updown = -1;

  ret = mg_get_http_var(&(hm->body), "interface", s_interface, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  ret = mg_get_http_var(&(hm->body), "state", s_updown, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  // only allow eth0 and up/down for now
  //
  if (strncmp(s_interface, "eth0", strlen("eth0")+1)!=0) {
    mg_http_send_error(nc, 404, NULL);
    return;
  }

  if (strncmp(s_updown, "up", strlen("up")+1)==0) {
    _iface_updown=1;
  }
  else if (strncmp(s_updown, "down", strlen("down")+1)==0) {
    _iface_updown=0;
  }

  if (_iface_updown < 0) {
    mg_http_send_error(nc, 404, NULL);
    return;
  }

  // Assuming we're running as root for now.
  // Bring eth0 interface up or down.
  //
  if (_iface_updown==1) {
    ret = system("/sbin/ip link set eth0 up &");
  }
  else if (_iface_updown==0) {
    ret = system("/sbin/ip link set eth0 down &");
  }

  if (ret!=0) {
    mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
        (unsigned long)strlen(msg_fail), msg_fail);
    return;
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_success), msg_success);
}


// get network interface information
//
static void ui_handle_interfaceinfo(struct mg_connection *nc, struct http_message *hm) {
  int n, fam, s;
  struct ifaddrs *_addr, *ifa;
  char host[NI_MAXHOST];

  char msg_fail[] = "fail interfaceinfo";
  char buf[4*_SLEN];

  // We're going to restrict to eth0 and only ip4
  //
  if (getifaddrs(&_addr)==-1) {
    mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
        (unsigned long)strlen(msg_fail), msg_fail);
    return;
  }

  for (ifa = _addr, n=0; ifa!=NULL; ifa = ifa->ifa_next, n++) {
    if (ifa->ifa_addr == NULL) { continue; }

    fam = ifa->ifa_addr->sa_family;
    if (fam != AF_INET) { continue; }
    if (strncmp(ifa->ifa_name, "eth0", strlen("eth0") + 1)!=0) { continue; }

    s = getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in),
        host, NI_MAXHOST,
        NULL, 0, NI_NUMERICHOST);
    if (s != 0) {
      mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
          (unsigned long)strlen(msg_fail), msg_fail);
      return;
    }
    snprintf(buf, (4*_SLEN) - 1, "ok msg=interfaceinfo\n%s=%s", ifa->ifa_name, host);
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(buf), buf);
}


// run 'say' custom script for voice text to speech
//
static void ui_handle_say(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char msg_ok[] = "ok say";

  char s_message[_SLEN];
  char buf[4*_SLEN];

  ret = mg_get_http_var(&(hm->body), "message", s_message, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  snprintf(buf, 4*_SLEN-1, "/home/bus/bin/say \"%s\" &", s_message);
  system(buf);

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_ok), msg_ok);
}

// change equipment number
//
static void ui_handle_eqnum(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char msg_ok[] = "ok eqnum";
  char s_eqnum[_SLEN];

  ret = mg_get_http_var(&(hm->body), "eqnum", s_eqnum, _SLEN);
  if (ret<=0) { mg_http_send_error(nc, 404, NULL); return; }

  ret = set_equip_num(atoi(s_eqnum));
  if (ret < 0) {
    mg_http_send_error(nc, 404, NULL);
    return;
  }

  mg_printf(nc, "HTTP/1.1 200 OK\r\nContent-Length: %lu\r\n\r\n%s",
      (unsigned long)strlen(msg_ok), msg_ok);
}


//-----
//-----
//-----

// api point
//
static void api_handle_req(struct mg_connection *nc, struct http_message *hm) {
  int ret;
  char s_action[_SLEN];

  ret = mg_get_http_var(&(hm->body), "action", s_action, _SLEN);

  if (ret==0) {
    mg_http_send_error(nc, 404, NULL);
    return;
  }

  if (strncmp(s_action, "driverlogin", strlen("driverlogin")+1)==0) {
    ui_handle_driver_login(nc, hm);
  }

  else if (strncmp(s_action, "paddleinput", strlen("paddleinput")+1)==0) {
    ui_handle_paddle_input(nc, hm);
  }

  else if (strncmp(s_action, "prevstop", strlen("prevstop")+1)==0) {
    ui_handle_prevstop_input(nc, hm);
  }

  else if (strncmp(s_action, "nextstop", strlen("nextstop")+1)==0) {
    ui_handle_nextstop_input(nc, hm);
  }

  else if (strncmp(s_action, "status", strlen("status")+1)==0) {
    ui_handle_status_input(nc, hm);
  }

  else if (strncmp(s_action, "logout", strlen("logout")+1)==0) {
    ui_handle_logout_input(nc, hm);
  }

  else if (strncmp(s_action, "fare", strlen("fare")+1)==0) {
    ui_handle_fare_input(nc, hm);
  }

  else if (strncmp(s_action, "dim", strlen("dim")+1)==0) {
    ui_handle_dim(nc, hm);
  }

  else if (strncmp(s_action, "volume", strlen("volume")+1)==0) {
    ui_handle_volume(nc, hm);
  }

  else if (strncmp(s_action, "interfaceupdown", strlen("interfaceupdown")+1)==0) {
    ui_handle_interfaceupdown(nc, hm);
  }

  else if (strncmp(s_action, "interfaceinfo", strlen("interfaceinfo")+1)==0) {
    ui_handle_interfaceinfo(nc, hm);
  }

  else if (strncmp(s_action, "say", strlen("say")+1)==0) {
    ui_handle_say(nc, hm);
  }

  else if (strncmp(s_action, "eqnum", strlen("eqnum")+1)==0) {
    ui_handle_eqnum(nc, hm);
  }

  else {
    mg_http_send_error(nc, 404, NULL);
  }

}

static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) {
  struct http_message *hm = (struct http_message *) ev_data;
  char buf[1024];

  int debug_print = 0;

  if (debug_print) {
    mg_sock_addr_to_str(&(nc->sa), buf, 1023, MG_SOCK_STRINGIFY_IP);
    printf("%s\n", buf);
  }

  switch (ev){

    case MG_EV_HTTP_REQUEST:

      if (mg_vcmp(&hm->uri, "/req")==0) {
        api_handle_req(nc, (struct http_message *)ev_data);
      }
      else {
        mg_serve_http(nc, (struct http_message *) ev_data, s_http_server_opts);
      }
      break;

    case MG_EV_WEBSOCKET_HANDSHAKE_DONE:
      break;
    case MG_EV_WEBSOCKET_FRAME:
      process_ws_message((struct websocket_message *)ev_data);
      break;

    case MG_EV_CLOSE:

      if (is_websocket(nc)) {
        // websocket closed
      }
      break;

    default:
      break;
  }

}


//DEBUG
void _test_ws(struct mg_mgr *mgr) {
  struct mg_connection *conn;
  char buf[_SLEN];

  snprintf(buf, _SLEN, "test message");

  for (conn = mg_next(mgr, NULL); conn; conn = mg_next(mgr, conn)) {
    if (!is_websocket(conn)) { continue; }
    mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT, buf, strlen(buf));
  }

}

//-------- web server functions ----


int main(int argc, char **argv) {
  char line[LINE_BUFFER_SIZE] = {0};

  struct message_record incoming_msg;
  struct message_record outgoing_msg;

  struct pollfd fds[32];
  int nfd=0;

  int poll_return;
  int read_return;

  int i;

  time_t down_time = 0;
  int calibration = 0;

  time_t now;

  int retval = 0;
  int tz = 0;
  int tx = 0;
  int ty = 0;

  time_t last_stale_gps_check = 0;

  long long int _usec_now, _usec_prv, _usec_del;
  _usec_now = get_usec_time();
  _usec_prv = _usec_now;
  _usec_del = 60000000;

  // setup mongoose web server
  //
  struct mg_connection *nc;

  mg_mgr_init(&g_mgr, NULL);
  nc = mg_bind(&g_mgr, s_http_port, ev_handler);
  if (!nc) {
    printf("failed to create listener\n");
    return 1;
  }

  mg_set_protocol_http_websocket(nc);
  s_http_server_opts.document_root = CONFIG_FILE_PATH "/html";
  s_http_server_opts.enable_directory_listing = "no";


  // Configure our signal handlers to deal with SIGINT, SIGTERM, etc... and make graceful exits while logging
  //
  configure_signal_handlers(argv[0]);

  // Make an initial attempt to get in touch with the interprocess communication hub (it may not be up yet depending on the start order)
  //
  maintain_ipc_hub_connect(argv[0]);

  // Register our defualt system message processing callbacks
  //
  register_system_status_callbacks();

  register_dispatch_callback(MAILBOX_STATUS_REQUEST, CALLBACK_USER(1), handle_status_request, NULL);
  register_dispatch_callback(MAILBOX_DRIVER_NOTIFY, CALLBACK_USER(2), handle_driver_notify, NULL);
  register_dispatch_callback(MAILBOX_PADDLE_ACK, CALLBACK_USER(3), handle_paddle_ack, NULL);
  register_dispatch_callback(MAILBOX_VAULT_DROP, CALLBACK_USER(4), handle_vault_drop, NULL);
  register_dispatch_callback(MAILBOX_TOKEN_MAG, CALLBACK_USER(5), handle_token_diag, NULL);
  register_dispatch_callback(MAILBOX_TOKEN_RFID, CALLBACK_USER(6), handle_token_diag, NULL);

  clear_diu_messages();

  // This is the main dispatch loop:
  //
  // * reset watchdog to make sure we haven't crashed/frozen
  // * if need be, open a connection to the DIU microcontroller, quieting all messages except for acks
  // * handle GPS message dispatch through IPC
  // * manage driver status message communication
  // * handle paddle change
  // * listen on the mailboxes for messages and process. This includes
  //   - gps updates
  //   - warning/debug/error messages
  //


  //loop until we get asked to exit.../
  //
  while( exit_request_status == EXIT_REQUEST_NONE ) {

    _usec_now = get_usec_time();
    if ((_usec_now - _usec_prv) > _usec_del) {
      //DEBUG
      printf("[%lli] diu_minder: heartbeat\n", get_usec_time());
      //DEBUG

      _usec_prv = _usec_now;
    }

    RESET_WATCHDOG();

    maintain_ipc_hub_connect(argv[0]);

    if(diu_fd < 0) {

      diu_fd = open_rs232_device(DRIVER_UI_PORT, USE_DEFAULT_BAUD, RS232_LINE);

      if(diu_fd < 0) {
        //fprintf(stderr, "Cannot open serial port %s for DIU!\n", DRIVER_UI_PORT);
      }
      else {
        write(diu_fd, "/Q:aK\r", 6);  //Turn on all messages except ACKs for sent commands
      }
    }

    now = time(NULL);

    //Every second, we want to update the DIU clock (even though it only shows minutes, this covers power events...)
    //
    if((now - last_diu_clock) > 0) {

      if(diu_fd) {
        set_diu_clock(diu_fd, now);
        last_diu_clock = now;
      }

    }

    //Every second we want to check to make sure that our GPS data have note gone stale...
    //
    if((now - last_stale_gps_check) > 0) {

      //   If the stale check results in an update to my_gps_stat, we must update any other
      // modules which may be tracking GPS status via the IPC hub.
      //
      if( handle_stale_gps_condition() > 0 ) {

        // If we have a connection to the IPC hub
        //
        if(hub_fd >= 0) {
          //Go and toss the data to any other modules who happen to care about GPS
          prepare_message(&outgoing_msg, MAILBOX_GPS_STATUS, &my_gps_stat, sizeof(my_gps_stat));
          send_message(hub_fd, &outgoing_msg);
        }
      }

      // Either way, remember that we did this stale check.
      //
      last_stale_gps_check = now;
    }

    if(hup_request_status) {

      hup_request_status = 0;

      // reserve ui if necssary ...
      //
    }

    //If it is time to send out a driver status update
    //
    if(update_driver_status && (hub_fd >= 0)) {

      //do so...
      //
      prepare_message(&outgoing_msg, MAILBOX_DRIVER_STATUS, &my_driver_status, sizeof(my_driver_status));
      send_message(hub_fd, &outgoing_msg);
      update_driver_status = 0;
    }

    check_paddle_request();

    //If we have to redraw the UI
    //if(redraw_flag && !calibration)
    //
    if(!calibration) {

      //Redraw the menu reflecting any changes from the last touchscreen input
      //or other stimulus
      //draw_menu(mt);
      //
      #ifdef TOUCHSCREEN_QUIET
        write(diu_fd, "/Q:t\r", 5);    //Un-Quiet the touch screen, now that we've responded to the user we want
      #endif                    //to hear if they have any further input to give us...

    }

    nfd = 0;

    if(hub_fd >= 0) {
      fds[nfd].fd = hub_fd;
      fds[nfd].events = POLLIN;
      fds[nfd].revents = 0;
      nfd++;
    }

    if(diu_fd >= 0) {
      fds[nfd].fd = diu_fd;
      fds[nfd].events = POLLIN;
      fds[nfd].revents = 0;
      nfd++;
    }

    // Add our websocket socket to the list of file descriptors
    // to select on.
    //
    for (nc = g_mgr.active_connections; nc != NULL; nc = nc->next) {

      if (nc->sock != INVALID_SOCKET) {
        if (nfd < 32) {
          fds[nfd].fd = nc->sock;
          fds[nfd].events = POLLIN | POLLOUT;
          fds[nfd].revents = 0;
          nfd++;
        }
      }
    }

    //if we have any file descriptors, poll them
    //
    if(nfd > 0) {
      poll_return = poll(fds, nfd, POLL_TIMEOUT);
    }

    //otherwise, whistle and look busy
    //
    else {
      // (this keeps us from buringing 100% cpu cycles if we don't have contact with either
      // the IPC hub or the DIU hardware).
      //
      poll_return = 0;
      sleep(1);
    }

    //--------------------------------------------------------------------------------------------------

    //if poll didn't net us any work to do,
    //
    if(poll_return < 1) {
      //lets try again
      //
      continue;
    }

    for(i = 0; i < nfd; i++)  //Loop through all polled file descriptors
    {

      //If we're looking at the DIU...
      //
      if( fds[i].fd == diu_fd ) {

        //if poll says our serial port has become bogus...
        //
        if(fds[i].revents & (POLLHUP | POLLERR | POLLNVAL)) {
          fprintf(stderr, "This is very odd... Poll returned flags %d on our serial port...\n", fds[i].revents);

          close(diu_fd);
          diu_fd = -1;
          break;
        }

        if(fds[i].revents & POLLIN) {
          read_return = read(fds[i].fd, line, sizeof(line));

          if(read_return > 0) {
            char *trav = line;

            line[read_return] = '\0';
            strip_crlf(line);

            //advance until EOL or we hit our start sentinel
            while(*trav && (*trav != '/') ) { trav++; }

            //Check to see that our address header is intact...
            //
            if( (trav[0] == '/') && (trav[2] == ':') ) {

              switch(trav[1]) {

                //-----------------------------------"/T:" means it's a touchscreen event
                //
                case 'T':

                  //advance past the header
                  //
                  trav += 3;

                  retval = sscanf(trav, "%x,%x,%x", &tz, &tx, &ty);

                  if(retval == 3) {
                    if(tz) {

                      if(down_time == 0) {
                        down_time = time(NULL);
                      }
                      else {
                        if( (time(NULL) - down_time) > TS_CALIB_HOLD_TIME) {
                          begin_touchscreen_calibration();
                          calibration = 1;
                        }
                      }


                      if(!calibration) {
                        // calibration from here turned off...
                      }
                      else {
                        if(advance_touchscreen_calibration(tx, ty, tz)) {
                          calibration = 0;
                        }
                        else {
                          draw_touchscreen_calibration();
                        }
                      }
                    }
                    else
                    {
                      down_time = 0;
                      if(!calibration) { }
                      else {
                        if(advance_touchscreen_calibration(tx, ty, tz)) {
                          calibration = 0;
                          //redraw_flag = 1;
                        }
                        else {
                          draw_touchscreen_calibration();
                        }
                      }
                    }
                  }

                  break;

                //-----------------------------------"/G:" means it is a new GPS input
                //
                case 'G':

                  //advance past the header
                  //
                  trav += 3;

                  //If this GPS update constitutes a meaningful piece of data
                  //
                  if(update_gps(trav) > 0)  {

                    //and we have a connection to the IPC hub
                    //
                    if(hub_fd >= 0) {

                      //Go and toss the data to any other modules who happen to care about GPS
                      //
                      prepare_message(&outgoing_msg, MAILBOX_GPS_STATUS, &my_gps_stat, sizeof(my_gps_stat));
                      send_message(hub_fd, &outgoing_msg);
                    }

                    last_stale_gps_check = now;  //Remember that we did a stale GPS check as part of our update.
                  }
                  break;

                //handle warnings
                //
                case '*':

                //debugs
                //
                case '#':

                //and errors
                //
                case '!':

                  //If this DIU error/debug message has not run afoul of the rate limiting policy...
                  //
                  if( can_report_diu_error() ) {

                    //send them all to the log server
                    //
                    format_log_message(&outgoing_msg, trav[1], "DIU Reports: %s", trav + 3);
                    send_message(hub_fd, &outgoing_msg);

                    //but in the case of errors, send them to the driver too
                    if(trav[1] == '!') {
                      format_driver_message(&outgoing_msg, trav[1], "DIU Reports: %s", trav + 3);
                      send_message(hub_fd, &outgoing_msg);
                    }
                  }
                  break;

                default:  //ignore any message addresses that we don't know what to do with
                  printf("Ignoring command \"%s\"\n", trav);
                  break;
              }
            }
            else {
              //printf("Ignoring non-command line \"%s\"\n", trav);
            }
          }
          else {
            fprintf(stderr, "Read from %s returned %d!\n", DRIVER_UI_PORT, read_return);

            close(diu_fd);
            diu_fd = -1;
            break;
          }
        }

      }

      //If we're looking at the IPC hub...
      //
      else if( fds[i].fd == hub_fd ) {

        //if poll says our connection to the IPC hub has died...
        //
        if(fds[i].revents & (POLLHUP | POLLERR | POLLNVAL)) {
          fprintf(stderr, "The connection to the IPC hub has gone away...\n");

          close(hub_fd);
          hub_fd = -1;
          break;
        }

        //if we have mail in any of our mailboxes...
        //
        if(fds[i].revents & POLLIN) {
          read_return = get_message(hub_fd, &incoming_msg);

          if(read_return < 0) {
            fprintf(stderr, "The connection to the IPC hub has gone away...\n");

            close(hub_fd);
            hub_fd = -1;
            break;
          }
          else {
            message_callback_return msg_status;
            msg_status = process_message(&incoming_msg);
            if (msg_status) {
              //pass
            }
          }

        }

      }
      else {

        if (fds[i].revents & (POLLIN | POLLOUT)) {
          mg_mgr_poll(&g_mgr, 10);
        }

      }
    }
  }

  if(hub_fd >= 0) {
    close(hub_fd);
  }

  if(diu_fd >= 0) {
    write(diu_fd, "/C:----\r",8);
    close(diu_fd);
  }

  return 0;
}