popufare/busunit/commhub/commhub.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 <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <poll.h>
#include <errno.h>
#include <string.h>

#include <sys/un.h>
#include <sys/time.h>
#include <sys/socket.h>

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

// This function gets the number of microseconds since epoch for storing in message headers and
// comparison to determine how timely a message is.
//
long long int get_usec_time() {
  struct timeval tv;

  // Get the number of seconds (and remainder microseconds) since epoch.
  //
  gettimeofday(&tv,NULL);

  // Return those in a single sane 64 bit integer containing the total number
  // of microseconds since epoch by multiplying tv_sec by a million and summing
  // the two values
  //
  return (long long int)tv.tv_usec + (1000000 * (long long int)tv.tv_sec);
}

// This function does a non-blocking poll to determine if the message passing socket is
// ready for input or output (depending on
// since it is much more efficient to block on the entire list of your I/O file descriptors
// and service any that cause you to unblock.
//
//  Returns:  MSG_NO_ACTION = no message
//      MSG_ERROR   = fatal error (passed fd is no good)
//
//      If the socket is ready, the return value will be
//      whatever combination of events specified in mask (MSG_SEND, MSG_RECV)
//      that is actually ready.
//
int message_socket_status(int message_socket_fd, int mask) {
  struct pollfd fds[1];
  int retval;

  fds[0].fd = message_socket_fd;

  do {
    fds[0].events = mask;
    retval = poll(fds,1,0);    //poll the 1 file descriptor, returning immediately.

    if (retval < 0) {

      // if we were interrupted by a signal, try again...
      //
      if(errno == EINTR) {
        continue;
      }

      // otherwise poll() has failed, which is BAD news.
      //
      return MSG_ERROR;
    }

    if (retval == 0) {
      return MSG_NO_ACTION;
    }

    // If poll tells us that the fd is invalid or has encountered a serious error, return -1.
    //
    if(fds[0].revents & (POLLERR | POLLNVAL)) {
      return MSG_ERROR;
    }

    // If any of the events supplied in mask have occurred, report them.
    //
    if(fds[0].revents & mask) {
      return fds[0].revents & mask;
    }

    // if there are no more message to drain (see above) and a hangup has occurred,
    // let the user know the socket is done for.
    if (fds[0].revents & POLLHUP) {
      return MSG_ERROR;
    }

    // Otherwise, by process of elimination we have no message and no error condition to report
    return MSG_NO_ACTION;

  } while(1);
}

// this function will pack a mailbox address and a payload into the supplied message structure and
// set the correct timestamp.
//
int prepare_message(struct message_record *target, char *mailbox, void *payload, unsigned int payload_len) {
  if (target == NULL) {
    return -1;
  }

  if (mailbox == NULL) {
    return -1;
  }

  if (payload == NULL) {
    return -1;
  }

  if (payload_len > MAX_PAYLOAD_LENGTH) {
    return -1;
  }

  memset(target, 0, sizeof(struct message_record));

  strncpy(target->header.mailbox_name, mailbox, MAILBOX_NAME_MAX);
  target->header.mailbox_name[MAILBOX_NAME_MAX] = '\0';
  target->header.usec_time = get_usec_time();
  target->header.payload_length = payload_len;
  target->header.sender = getpid();
  target->header.from_fd = 0;

  memmove(target->payload, payload, payload_len);

  return 0;
}

// the send_message function will transmit a message and its header (trimmed to the specified payload length).
// zero will be returned on success, -1 on failure.
//
int send_message(int message_socket_fd, struct message_record *message) {
  int txlen;
  int retval;

  if (message == NULL) {
    return -1;
  }

  // If we are trying to send a message that is longer than allowed, throw an error!
  //
  if (message->header.payload_length > MAX_PAYLOAD_LENGTH) {
    return -1;
  }

  // Calculate the total message length (header + payload) so that we don't send
  // any extra padding that we don't need.
  txlen = sizeof(struct message_header_record) + message->header.payload_length;

  do {
    retval = send(message_socket_fd, message, txlen, MSG_EOR);

    // if we were interrupted by a signal, try again.
    //
    if ( (retval == -1) && (errno == EINTR) ) {
      continue;
    }

    // if we tried to send a message and it got truncated, return failure.
    //
    if (retval != txlen) {
      return -1;
    }

    return 0;

  } while(1);
}

// the get_message function will receive a message from the opposite end of the socket and place it in the
// supplied message structure incliding its header.  -1 is returned on failure, 0 on success.
//
int get_message(int message_socket_fd, struct message_record *message) {
  int retval;

  if (message == NULL) {
    return -1;
  }

  // clear our buffer of any garbage before using it to read in the message
  //
  memset(message, 0, sizeof(struct message_record));

  do {
    // try and receive our message
    //
    retval = recv(message_socket_fd, message, sizeof(struct message_record), 0);

    // if we were interrupted by a signal, try again.
    //
    if ( (retval == -1) && (errno == EINTR) ) {
      continue;
    }

    // zero is a special case for recv which indicates that the other side has shut down the connection.
    if (retval == 0) {
      return -1;
    }

    // if the received message size does not add up (between header and payload)
    //
    if (retval != (sizeof(struct message_header_record) + message->header.payload_length)) {

      // throw an error!
      //
      return -1;
    }

    // Record which IPC socket we got this from (for things like PING replies)
    message->header.from_fd = message_socket_fd;

    // otherwise, return success
    //
    return 0;

  } while(1);
}

// This function connects to the communication hub and returns the file descriptor of the connection.
// If the connection failed, -1 is returned.  The progname parameter (if not NULL) will be used to
// register a module name (snagged from argv[0]) with the server for ease of debugging.
//
int connect_to_message_server(char *progname) {
  int fd;
  int retval;
  int len;
  char *message_text;
  struct sockaddr_un addr;

  struct message_record msg;

  fd = socket(AF_UNIX, SOCK_SEQPACKET, 0);

  if (fd < 0) {
    return -1;
  }

  // construct our address structure (pointing to the agreed upon server address)
  // to pass to connect.
  addr.sun_family = AF_UNIX;
  strncpy(addr.sun_path, COMMHUB_ADDRESS, sizeof(addr.sun_path) - 1);
  addr.sun_path[sizeof(addr.sun_path) - 1]='\0';
  len=strlen(COMMHUB_ADDRESS) + sizeof(addr.sun_family);

  // attempt to connect to said server
  //
  retval = connect(fd,(struct sockaddr *)&addr,len);

  if (retval) {
    fprintf(stderr, "Cannot connect to IPC hub at %s\n", COMMHUB_ADDRESS);
    close(fd);
    return -1;
  }

  // if we have been passed a module name to register as, use that, otherwise default to "ANONYMOUS"
  //
  if (progname) {
    message_text = progname;
  }

  else {
    message_text = "ANONYMOUS";
  }

  len = strlen(message_text);

  // if for some inconceivable reason our module name is longer than the server allows, truncate it.
  //
  if (len > MAX_MODULE_NAME_LENGTH) {
    len = MAX_MODULE_NAME_LENGTH;
  }

  // prepare and send our registation message
  //
  prepare_message(&msg, MAILBOX_HELLO, message_text, len);
  send_message(fd, &msg);

  return fd;
}

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


// Read the `state_info_t` from the on disk text file.
// Format is field (in caps) followed by the field value
// separated by a single space.
//
// return:
// 0 on success
// non-zero on error
//
int get_state_info(state_info_t *_state) {
  FILE *fp;
  int ch = 1,
      input_idx=0,
      n=0,
      field_len=0;
  char buffer[LINE_BUFFER_SIZE],
       *chp;

  if (access(STATE_INFO_FILE, R_OK)!=0) {
    return -1;
  }

  memset(buffer, 0, sizeof(char)*LINE_BUFFER_SIZE);

  if ((fp = fopen(STATE_INFO_FILE, "r")) == NULL) {
    return -1;
  }

  while (!feof(fp)) {
    ch = fgetc(fp);
    if ((ch == '\n') || (ch == EOF)) {

      // ignore blank lines and comment lines
      // (begining with '#')
      //
      if (input_idx == 0) { continue; }
      if (buffer[0] == '#') {
        buffer[0] = '\0';
        input_idx = 0;
        continue;
      }

      chp = strchr(buffer, ' ');
      if (chp != NULL) {
        chp++;

        n = chp - buffer;

        if (strncmp(buffer, "LAT ", n)==0) {
          _state->lat = atof(chp);
        }

        else if (strncmp(buffer, "LON ", n)==0) {
          _state->lon = atof(chp);
        }

        else if (strncmp(buffer, "HEADING ", n)==0) {
          _state->heading = atof(chp);
        }

        else if (strncmp(buffer, "VELOCITY ", n)==0) {
          _state->velocity = atof(chp);
        }

        else if (strncmp(buffer, "NUM_SATS ", n)==0) {
          _state->num_sats = atoi(chp);
        }

        else if (strncmp(buffer, "GPS_GOOD ", n)==0) {
          _state->gps_good = atoi(chp);
        }

        else if (strncmp(buffer, "STAMP ", n)==0) {
          _state->stamp = (time_t)atoi(chp);
        }

        else if (strncmp(buffer, "GPSTAMP ", n)==0) {
          _state->gpstime = (time_t)atoi(chp);
        }

        else if (strncmp(buffer, "PADDLE ", n)==0) {
          _state->paddle = atoi(chp);
        }

        else if (strncmp(buffer, "ROUTE ", n)==0) {
          _state->route = atoi(chp);
        }

        else if (strncmp(buffer, "TRIP ", n)==0) {
          _state->trip = atoi(chp);
        }

        else if (strncmp(buffer, "STOP ", n)==0) {
          _state->stop = atoi(chp);
        }

        else if (strncmp(buffer, "STOPNAME ", n)==0) {
          field_len = input_idx - n;
          if (field_len >= STATE_INFO_FIELD_SIZE) {
            field_len = STATE_INFO_FIELD_SIZE-1;
          }
          memcpy(_state->stopname, chp, field_len);
          _state->stopname[field_len] = '\0';
        }

        else if (strncmp(buffer, "LOGGED_IN_DRIVER ", n)==0) {
          _state->logged_in_driver = atoi(chp);
        }

        else if (strncmp(buffer, "DRIVER_NAME ", n)==0) {
          field_len = input_idx - n;
          if (field_len >= STATE_INFO_FIELD_SIZE) {
            field_len = STATE_INFO_FIELD_SIZE-1;
          }
          memcpy(_state->driver_name, chp, input_idx - n + 1);
          _state->driver_name[field_len] = '\0';
        }

        else if (strncmp(buffer, "EQUIP_NUM ", n)==0) {
          _state->equip_num = atoi(chp);
        }

      }

      buffer[0] = '\0';
      input_idx = 0;
      continue;
    }

    buffer[input_idx] = ch;
    input_idx++;
    if (input_idx >= LINE_BUFFER_SIZE) {
      input_idx = LINE_BUFFER_SIZE-1;
    }
    buffer[input_idx] = '\0';

  }

  fclose(fp);

  return 0;
}

// Save `state_info_t` on disk.
// Format it field name (in caps) separated by
// a single space followed by the field value.
//
// Constructs the text state file as a temporary file
// (specified by STATE_INFO_TEMPFILE) then moves it
// after it's finished.
//
// return:
// 0 on success
// non-zero on error
//
int set_state_info(state_info_t *_state) {
  FILE *fp;

  if ((fp = fopen(STATE_INFO_TEMPFILE, "w")) == NULL) {
    return -1;
  }

  fprintf(fp, "# updated %lli\n", get_usec_time());
  fprintf(fp, "# %s\n", _state->comment);

  fprintf(fp, "LAT %f\n", _state->lat);
  fprintf(fp, "LON %f\n", _state->lon);

  fprintf(fp, "HEADING %f\n", _state->heading);
  fprintf(fp, "VELOCITY %f\n", _state->velocity);

  fprintf(fp, "NUM_SATS %i\n", _state->num_sats);
  fprintf(fp, "GPS_GOOD %i\n", _state->gps_good);

  fprintf(fp, "STAMP %i\n", (int)(_state->stamp));
  fprintf(fp, "GPSTAMP %i\n", (int)(_state->gpstime));

  fprintf(fp, "PADDLE %i\n", _state->paddle);
  fprintf(fp, "ROUTE %i\n", _state->route);
  fprintf(fp, "TRIP %i\n", _state->trip);
  fprintf(fp, "STOP %i\n", _state->stop);

  fprintf(fp, "STOPNAME %s\n", _state->stopname);

  fprintf(fp, "LOGGED_IN_DRIVER %i\n", _state->logged_in_driver);
  fprintf(fp, "DRIVER_NAME %s\n", _state->driver_name);
  fprintf(fp, "EQUIP_NUM %i\n", _state->equip_num);

  fclose(fp);

  return rename(STATE_INFO_TEMPFILE, STATE_INFO_FILE);
}

// print out the state information
int print_state_info(state_info_t *_state) {
  FILE *fp = stdout;

  fprintf(fp, "LAT %f\n", _state->lat);
  fprintf(fp, "LON %f\n", _state->lon);

  fprintf(fp, "HEADING %f\n", _state->heading);
  fprintf(fp, "VELOCITY %f\n", _state->velocity);

  fprintf(fp, "NUM_SATS %i\n", _state->num_sats);
  fprintf(fp, "GPS_GOOD %i\n", _state->gps_good);

  fprintf(fp, "STAMP %i\n", (int)(_state->stamp));
  fprintf(fp, "GPSTAMP %i\n", (int)(_state->gpstime));

  fprintf(fp, "PADDLE %i\n", _state->paddle);
  fprintf(fp, "ROUTE %i\n", _state->route);
  fprintf(fp, "TRIP %i\n", _state->trip);
  fprintf(fp, "STOP %i\n", _state->stop);

  fprintf(fp, "STOPNAME %s\n", _state->stopname);

  fprintf(fp, "LOGGED_IN_DRIVER %i\n", _state->logged_in_driver);
  fprintf(fp, "DRIVER_NAME %s\n", _state->driver_name);
  fprintf(fp, "EQUIP_NUM %i\n", _state->equip_num);

  return 0;
}