popufare/busunit/avls/avls_communication.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/socket.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>

#include <netinet/in.h>
#include <arpa/inet.h>

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <poll.h>
#include <time.h>

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

//----------GLOBAL STATE VARIABLES

time_t   last_sync_attempt = 0;  //Time of the last normal chirp in seconds since epoch
time_t  last_offline_chirp = 0; //Time of the last offline chirp in seconds since epoch

int   commhub_fd = -1;  //File descriptor of our connection to the comm hub
int  server_fd = -1;    //File descriptor of our connection to the sync server

int  avls_motion_interval = AVLS_MOTION_INTERVAL;
int  avls_still_interval = AVLS_STILL_INTERVAL;

int load_avls_config()
{
    int retval;
    FILE *f;
    char line[LINE_BUFFER_SIZE];

    int still = 0, motion = 0;

    f = fopen(AVLS_CONFIG_FILE, "rb");

    if(f)
    {
        while( !feof(f) && fgets(line, LINE_BUFFER_SIZE - 1, f) )
        {
            retval = sscanf(line, "%d %d", &motion, &still);

            if(retval == 2)
            {
                avls_motion_interval = motion;
                avls_still_interval = still;

                //DEBUG
                fprintf(stderr, "#avls (motion %i, still %i) interval\n", avls_motion_interval, avls_still_interval);
                //DEBUG

                fclose(f);
                return 0;
            }
        }

        fclose(f);
    }

    return -1;
}

//This function attempts to connect to the pass server...
int connect_to_avls_server()
{
    int fd;
    int retval;
    struct sockaddr_in addr;

    fd = socket(PF_INET, SOCK_STREAM, 0);

    if(fd < 0)
        return -1;

    addr.sin_family = AF_INET;
    addr.sin_port = htons(AVLS_SERVER_PORT);
    inet_aton(AVLS_SERVER_IP, &addr.sin_addr);

    retval = connect(fd, (struct sockaddr *)&addr, sizeof(addr));

    if(retval < 0)
    {
        close(fd);
        return -2;
    }

  return fd;
}

int construct_avls_chirp(char *chirp, int size)
{
    int len = 0;

    len += snprintf(chirp + len, size - len, "%d\t", get_equip_num());
    len += snprintf(chirp + len, size - len, "%d\t", driver_stat.logged_in_driver);

    len += snprintf(chirp + len, size - len, "%d\t", stop_stat.paddle);
    len += snprintf(chirp + len, size - len, "%d\t", stop_stat.route);
    len += snprintf(chirp + len, size - len, "%d\t", stop_stat.trip);
    len += snprintf(chirp + len, size - len, "%d\t", stop_stat.stop);

    len += snprintf(chirp + len, size - len, "%d\t", (int)time(NULL));

    len += snprintf(chirp + len, size - len, "%f\t", gps_stat.lat);
    len += snprintf(chirp + len, size - len, "%f\t", gps_stat.lon);
    len += snprintf(chirp + len, size - len, "%f\t", gps_stat.heading);
    len += snprintf(chirp + len, size - len, "%f\n", gps_stat.velocity);

    return len;
}

int send_avls_chirp(int fd)
{
    char chirp[1024] = {0};
    int len = 0;

    len = construct_avls_chirp(chirp, sizeof(chirp));

    if(send(fd, chirp, len, 0) != len)
    {
        return -1;
    }

    return 0;
}

int send_offline_chirp()
{
    char chirp[1024] = {0};
    int len = 0;
    struct message_record outgoing_msg;

    len = construct_avls_chirp(chirp, sizeof(chirp));

    if(len >= sizeof(chirp))
    {
        chirp[sizeof(chirp) - 1] = '\0';
    }
    else
    {
        chirp[len]='\0';
    }

    format_log_message(&outgoing_msg, LOGLEVEL_DEBUG, "OFFLINE AVLS: %s", chirp);

    if(commhub_fd >= 0)
    {
        return send_message(commhub_fd, &outgoing_msg);
    }

    return -1;
}

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

    if(commhub_fd < 0)  //if we have no connection to the communication hub
    {
        commhub_fd = connect_to_message_server(progname);  //try and get one

//        printf("commhub_fd = %d\n", commhub_fd);

        if(commhub_fd >= 0)  //if it worked
        {
            //Subscribe to the relevant status management mailboxes
            subscribe_to_default_messages(commhub_fd);

            //Request updated status information...
            prepare_message(&outgoing_msg, MAILBOX_STATUS_REQUEST, "", 0);
            send_message(commhub_fd,&outgoing_msg);
        }
    }
}

int main(int argc, char **argv)
{
    struct pollfd fds[2];
    int nfds = 0;
    int poll_return = 0;
    int read_return = 0;

    int i;

    struct message_record incoming_msg;

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

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

    load_avls_config();

    configure_signal_handlers(argv[0]);

    maintain_ipc_hub_connect(argv[0]);

    //Register our default keep-up-with-system status callbacks
    register_system_status_callbacks();

    while( exit_request_status == EXIT_REQUEST_NONE )
    {
        time_t now = time(NULL);
        int sync_threshold = 0;

        RESET_WATCHDOG();

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

        if(hup_request_status)
        {
            load_avls_config();
            hup_request_status = 0;
        }

        maintain_ipc_hub_connect(argv[0]);

        if(server_fd < 0)  //If we don't have a connection to the sync server...
        {
            if( (now - last_sync_attempt) > DEFAULT_CONNECT_RETRY )  //See if it is time to try again
            {
                if( tunnel_is_up() )          //and if the tunnel thinks it is up
                {
                     server_fd = connect_to_avls_server();    //if so, try again...

//                     printf("server_fd = %d\n", server_fd);

                     if(server_fd >= 0)  //if it worked
                     {
                         last_sync_attempt = 0;
                     }
                     else
                     {
                         last_sync_attempt = now;
                     }
                }
            }
        }

        nfds=0;

        //-----------------------------------------------
        //
        //   Figure out what our time threshold is to transmit
        // a message based on whether we're moving or still.
        //
        //-----------------------------------------------
        if(gps_stat.velocity > MOTION_THRESHOLD)  //See if the bus is in motion...
        {
            sync_threshold = avls_motion_interval;  //if so, we want to send AVLS chirps at higher rate
        }
        else
        {
            sync_threshold = avls_still_interval;  //if not, we send them at a lower default rate
        }

        if(server_fd >= 0)  //---------- If we have an active connection to the AVLS server through the tunnel
        {
            fds[nfds].fd = server_fd;  //Add it to the list of things we must poll()
            fds[nfds].events = 0;

            // See if we have gone long enough to need to transmit a message AND have valid GPS data to transmit
            if( ((now - last_sync_attempt) > sync_threshold) && gps_stat.gps_good )  
            {
                fds[nfds].events |= POLLOUT;      //If so, ask poll to see if we have buffer space to transmit one
            }

            nfds++;            //either way, add this FD to the poll() list.
        }
        else      //---------- Otherwise, if we have no connection to the AVLS server
        {
            // See if it is time to transmit an offline chirp AND we have valid GPS data to transmit
            if( ((now - last_offline_chirp) > sync_threshold)  && gps_stat.gps_good )  
            {
                if(send_offline_chirp() == 0)      //If so, stick one in the diagnostic log at DEBUG priority
                {
                    last_offline_chirp = now;          //Remember that we've done so if it worked.
                }
            }
        }

        if(commhub_fd >= 0)
        {
            fds[nfds].fd = commhub_fd;
            fds[nfds].events = POLLIN;

            nfds++;
        }

        if(nfds > 0)
        {
            poll_return = poll(fds, nfds, POLL_TIMEOUT);
        }
        else
        {
            usleep(POLL_TIMEOUT * 1000);
            poll_return = 0;
        }

//        printf("Poll returns %d (of %d)\n", poll_return, nfds);
//        for(i=0; i < nfds; i++)
//        {
//            printf("\t%d: fd=%d ev=%d rev=%d\n", i, fds[i].fd, fds[i].events, fds[i].revents);
//        }

        if(poll_return <= 0)
        {
            continue;
        }

        for(i=0; i < nfds; i++)
        {
//            printf("Processing Poll Result %d\n",i);
            if( fds[i].fd == server_fd )
            {
                //If we've lost connection, break this loop and poll all over again
                if(fds[i].revents & (POLLERR | POLLHUP | POLLNVAL))
                {
                    close(server_fd);
                    server_fd = -1;
                    break;
                }

                if(fds[i].revents & POLLOUT)
                {
                    //sent a query here...

//                    printf("Trying to write to server...\n");
                    read_return = send_avls_chirp(server_fd);

                    //and then update our last sync attempt time
                    last_sync_attempt = now;

                    if(read_return < 0)
                    {
                        close(server_fd);
                        server_fd = -1;
                        break;
                    }

                }

            }
            else if( fds[i].fd == commhub_fd )
            {  
                //If we've lost connection, break this loop and poll all over again
                if(fds[i].revents & (POLLERR | POLLHUP | POLLNVAL))  
                {
                    close(commhub_fd);
                    commhub_fd = -1;
                    break;
                }

                if(fds[i].revents & POLLIN)
                {
//                    printf("Trying to read from hub...\n");
                    read_return = get_message(commhub_fd, &incoming_msg);

                    if( read_return < 0 )
                    {
                        close(commhub_fd);
                        commhub_fd = -1;
                        break;
                    }

                    process_message(&incoming_msg);  //This passes the received message through the callback list

                }

            }
        }
    }



    return 0;
}