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))
        {
            return -1;	//throw an error!
        }
 
        //Record which IPC socket we got this from (for things like PING replies)
        message->header.from_fd = message_socket_fd;
    
        return 0;	//otherwise, return success
          
    } 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;
}

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