upstream.c

/**
 * Copyright 2015 Comcast Cable Communications Management, LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
/**
 * @file upstream.c
 *
 * @description This describes functions required to manage upstream messages.
 *
 */

#include "ParodusInternal.h"
#include "upstream.h"
#include "config.h"
#include "partners_check.h"
#include "connection.h"
#include "client_list.h"
#include "nopoll_helpers.h"
#include "close_retry.h"

/*----------------------------------------------------------------------------*/
/*                                   Macros                                   */
/*----------------------------------------------------------------------------*/
#define METADATA_COUNT 					12
#define PARODUS_SERVICE_NAME			"parodus"
/*----------------------------------------------------------------------------*/
/*                            File Scoped Variables                           */
/*----------------------------------------------------------------------------*/

void *metadataPack;
size_t metaPackSize=0;


UpStreamMsg *UpStreamMsgQ = NULL;

pthread_mutex_t nano_mut=PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t nano_con=PTHREAD_COND_INITIALIZER;

UpStreamMsg * get_global_UpStreamMsgQ(void)
{
    return UpStreamMsgQ;
}

void set_global_UpStreamMsgQ(UpStreamMsg * UpStreamQ)
{
    UpStreamMsgQ = UpStreamQ;
}

pthread_cond_t *get_global_nano_con(void)
{
    return &nano_con;
}

pthread_mutex_t *get_global_nano_mut(void)
{
    return &nano_mut;
}

/*----------------------------------------------------------------------------*/
/*                             Internal Functions                             */
/*----------------------------------------------------------------------------*/

/*----------------------------------------------------------------------------*/
/*                             External functions                             */
/*----------------------------------------------------------------------------*/

void packMetaData()
{
    char boot_time[256]={'\0'};
    //Pack the metadata initially to reuse for every upstream msg sending to server
    ParodusPrint("-------------- Packing metadata ----------------\n");
    sprintf(boot_time, "%d", get_parodus_cfg()->boot_time);
    struct data meta_pack[METADATA_COUNT] = {
            {HW_MODELNAME, get_parodus_cfg()->hw_model},
            {HW_SERIALNUMBER, get_parodus_cfg()->hw_serial_number},
            {HW_MANUFACTURER, get_parodus_cfg()->hw_manufacturer},
            {HW_DEVICEMAC, get_parodus_cfg()->hw_mac},
            {HW_LAST_REBOOT_REASON, get_parodus_cfg()->hw_last_reboot_reason},
            {FIRMWARE_NAME , get_parodus_cfg()->fw_name},
            {BOOT_TIME, boot_time},
            {LAST_RECONNECT_REASON, get_global_reconnect_reason()},
            {WEBPA_PROTOCOL, get_parodus_cfg()->webpa_protocol},
            {WEBPA_UUID,get_parodus_cfg()->webpa_uuid},
            {WEBPA_INTERFACE, getWebpaInterface()},
            {PARTNER_ID, get_parodus_cfg()->partner_id}
        };
    const data_t metapack = {METADATA_COUNT, meta_pack};

    metaPackSize = wrp_pack_metadata( &metapack , &metadataPack );

    if (metaPackSize > 0) 
    {
	    ParodusInfo("metadata encoding is successful with size %zu\n", metaPackSize);
    }
    else
    {
	    ParodusError("Failed to encode metadata\n");
    }
}

void clear_metadata(){
    if(metadataPack != NULL)
         free(metadataPack);
}
       
/*
 * @brief To handle UpStream messages which is received from nanomsg server socket
 */

void *handle_upstream()
{
    UpStreamMsg *message;
    int sock, bind;
    int bytes =0;
    void *buf;

    ParodusPrint("******** Start of handle_upstream ********\n");

    sock = nn_socket( AF_SP, NN_PULL );
    if(sock >= 0)
    {
        int t = NANO_SOCKET_RCV_TIMEOUT_MS;
        int rc = nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &t, sizeof(t));
        if (rc < 0)
        {
            ParodusError ("Unable to set socket receive timeout (errno=%d, %s)\n",errno, strerror(errno));
        }
        ParodusPrint("Nanomsg bind with get_parodus_cfg()->local_url  %s\n", get_parodus_cfg()->local_url);
        bind = nn_bind(sock, get_parodus_cfg()->local_url);
        if(bind < 0)
        {
            ParodusError("Unable to bind socket (errno=%d, %s)\n",errno, strerror(errno));
        }
        else
        {
            ParodusInfo("nanomsg server gone into the listening mode...\n");
            while( FOREVER() ) 
            {
                buf = NULL;
                bytes = nn_recv (sock, &buf, NN_MSG, 0);
                if (g_shutdown)
                  break;
                if (bytes < 0) {
                  if ((errno != EAGAIN) && (errno != ETIMEDOUT))
                    ParodusInfo ("Error (%d) receiving message from nanomsg client\n", errno);
                  continue;
                }
                ParodusInfo ("Upstream message received from nanomsg client\n");
                message = (UpStreamMsg *)malloc(sizeof(UpStreamMsg));

                if(message)
                {
                    message->msg =buf;
                    message->len =bytes;
                    message->next=NULL;
                    pthread_mutex_lock (&nano_mut);
                    //Producer adds the nanoMsg into queue
                    if(UpStreamMsgQ == NULL)
                    {
                        UpStreamMsgQ = message;

                        ParodusPrint("Producer added message\n");
                        pthread_cond_signal(&nano_con);
                        pthread_mutex_unlock (&nano_mut);
                        ParodusPrint("mutex unlock in producer thread\n");
                    }
                    else
                    {
                        UpStreamMsg *temp = UpStreamMsgQ;
                        while(temp->next)
                        {
                            temp = temp->next;
                        }
                        temp->next = message;
                        pthread_mutex_unlock (&nano_mut);
                    }
                }
                else
                {
                    ParodusError("failure in allocation for message\n");
                }
            }
	    if(nn_shutdown(sock, bind) < 0)
	    {
	        ParodusError ("nn_shutdown bind socket=%d endpt=%d, err=%d\n", 
		    sock, bind, errno);
	    }
	    if (nn_close (sock) < 0)
	    {
	        ParodusError ("nn_close bind socket=%d err=%d\n", 
		    sock, errno);
	    }
        }
    }
    else
    {
        ParodusError("Unable to create socket (errno=%d, %s)\n",errno, strerror(errno));
    }
    ParodusPrint ("End of handle_upstream\n");
    return 0;
}


void *processUpstreamMessage()
{		
    int rv=-1, rc = -1;	
    int msgType;
    wrp_msg_t *msg,*retrieve_msg = NULL;
    void *bytes;
    reg_list_item_t *temp = NULL;
    int matchFlag = 0;
    int status = -1;
	char *device_id = NULL;
	size_t device_id_len = 0;
	size_t parodus_len;
	int ret = -1;

    while(FOREVER())
    {
        pthread_mutex_lock (&nano_mut);
        ParodusPrint("mutex lock in consumer thread\n");
        if(UpStreamMsgQ != NULL)
        {
            UpStreamMsg *message = UpStreamMsgQ;
            UpStreamMsgQ = UpStreamMsgQ->next;
            pthread_mutex_unlock (&nano_mut);
            ParodusPrint("mutex unlock in consumer thread\n");

            /*** Decoding Upstream Msg to check msgType ***/
            /*** For MsgType 9 Perform Nanomsg client Registration else Send to server ***/	
            ParodusPrint("---- Decoding Upstream Msg ----\n");

            rv = wrp_to_struct( message->msg, message->len, WRP_BYTES, &msg );
            if(rv > 0)
            {
                msgType = msg->msg_type;				   
                if(msgType == WRP_MSG_TYPE__SVC_REGISTRATION)
                {
                    ParodusInfo("\n Nanomsg client Registration for Upstream\n");
                    //Extract serviceName and url & store it in a linked list for reg_clients
                    temp = get_global_node();
                    if(get_numOfClients() !=0)
                    {
                        matchFlag = 0;
                        ParodusPrint("matchFlag reset to %d\n", matchFlag);
                        while(temp!=NULL)
                        {
                            if(strcmp(temp->service_name, msg->u.reg.service_name)==0)
                            {
                                ParodusInfo("match found, client is already registered\n");
                                parStrncpy(temp->url,msg->u.reg.url, sizeof(temp->url));
                                if(nn_shutdown(temp->sock, temp->endpoint) < 0)
                                {
                                    ParodusError ("nn_shutdown socket=%d endpt=%d, err=%d\n", 
					temp->sock, temp->endpoint, errno);
                                }
				if (nn_close (temp->sock) < 0)
                                {
                                    ParodusError ("nn_close socket=%d err=%d\n", 
					temp->sock, errno);
                                }

                                temp->sock = nn_socket(AF_SP,NN_PUSH );
                                if(temp->sock >= 0)
                                {					
                                    int t = NANO_SOCKET_SEND_TIMEOUT_MS;
                                    rc = nn_setsockopt(temp->sock, NN_SOL_SOCKET, NN_SNDTIMEO, &t, sizeof(t));
                                    if(rc < 0)
                                    {
                                        ParodusError ("Unable to set socket send timeout (errno=%d, %s)\n",errno, strerror(errno));
                                    }
                                    rc = nn_connect(temp->sock, msg->u.reg.url); 
                                    if(rc < 0)
                                    {
                                        ParodusError ("Unable to connect socket (errno=%d, %s)\n",errno, strerror(errno));
                                    }
                                    else
                                    {
                                        temp->endpoint = rc;
                                        ParodusInfo("Client registered before. Sending ack on socket %d\n", temp->sock); 
                                        status =sendAuthStatus(temp);

                                        if(status == 0)
                                        {
                                            ParodusPrint("sent auth status to reg client\n");
                                        }
                                        matchFlag = 1;
                                        break;
                                    }
                                }
                                else
                                {
                                    ParodusError("Unable to create socket (errno=%d, %s)\n",errno, strerror(errno));
                                }
                            }
                            ParodusPrint("checking the next item in the list\n");
                            temp= temp->next;
                        }	
                    }
                    ParodusPrint("matchFlag is :%d\n", matchFlag);
                    if((matchFlag == 0) || (get_numOfClients() == 0))
                    {
                        ParodusPrint("Adding nanomsg clients to list\n");
                        status = addToList(&msg);
                        ParodusPrint("addToList status is :%d\n", status);
                        if(status == 0)
                        {
                            ParodusPrint("sent auth status to reg client\n");
                        }
                    }
                    release_global_node ();
                }
                else if(msgType == WRP_MSG_TYPE__EVENT)
                {
                    (msg->u.event.headers != NULL && msg->u.event.headers->headers[0] != NULL && msg->u.event.headers->headers[1] != NULL) ? ParodusInfo(" Received upstream event data: dest '%s' traceParent: %s traceState: %s\n", msg->u.event.dest, msg->u.event.headers->headers[0], msg->u.event.headers->headers[1]) : ParodusInfo(" Received upstream event data: dest '%s'\n", msg->u.event.dest);
		    if(msg->u.event.transaction_uuid != NULL) {
			    ParodusInfo("transaction_uuid in event: %s\n", msg->u.event.transaction_uuid);
		    }	    
                    partners_t *partnersList = NULL;
                    int j = 0;

                    int ret = validate_partner_id(msg, &partnersList);
                    if(ret == 1)
                    {
                        wrp_msg_t *eventMsg = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
			memset( eventMsg, 0, sizeof( wrp_msg_t ) );
                        eventMsg->msg_type = msgType;
                        eventMsg->u.event.content_type=msg->u.event.content_type;
                        eventMsg->u.event.source=msg->u.event.source;
                        eventMsg->u.event.dest=msg->u.event.dest;
                        eventMsg->u.event.payload=msg->u.event.payload;
                        eventMsg->u.event.payload_size=msg->u.event.payload_size;
                        eventMsg->u.event.headers=msg->u.event.headers;
                        eventMsg->u.event.metadata=msg->u.event.metadata;
                        eventMsg->u.event.partner_ids = partnersList;
			if(msg->u.event.transaction_uuid)
			{
				ParodusPrint("Inside Trans id in PARODUS\n");
			}
			else
			{
				ParodusPrint("Assigning NULL to trans id\n");
				eventMsg->u.event.transaction_uuid = NULL;
			}

                        int size = wrp_struct_to( eventMsg, WRP_BYTES, &bytes );
                        if(size > 0)
                        {
                            sendUpstreamMsgToServer(&bytes, size);
                        }
                        free(eventMsg);
                        free(bytes);
                        bytes = NULL;
                    }
                    else
                    {
                        sendUpstreamMsgToServer(&message->msg, message->len);
                    }
                    if(partnersList != NULL)
                    {
                        for(j=0; j<(int)partnersList->count; j++)
                        {
                            if(NULL != partnersList->partner_ids[j])
                            {
                                free(partnersList->partner_ids[j]);
                            }
                        }
                        free(partnersList);
                    }
                    partnersList = NULL;
                }
                else
                {
					//Sending to server for msgTypes 3, 5, 6, 7, 8.
					if( WRP_MSG_TYPE__REQ == msgType )
					{
						(msg->u.req.headers != NULL && msg->u.req.headers->headers[0] != NULL && msg->u.req.headers->headers[1] != NULL) ? ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s traceParent: %s traceState: %s\n", msgType, msg->u.req.dest, msg->u.req.transaction_uuid, msg->u.req.headers->headers[0], msg->u.req.headers->headers[1]) : ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s\n", msgType, msg->u.req.dest, msg->u.req.transaction_uuid);
						sendUpstreamMsgToServer(&message->msg, message->len);
					}
					else
					{
						ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s status: %d\n",msgType, msg->u.crud.dest, msg->u.crud.transaction_uuid, msg->u.crud.status );
						if(WRP_MSG_TYPE__RETREIVE == msgType)
						{
								ret = getDeviceId(&device_id, &device_id_len);
								if(ret == 0)
								{
									ParodusPrint("device_id %s device_id_len %lu\n", device_id, device_id_len);
									/* Match dest based on device_id. Check dest start with: "mac:112233445xxx" ? */
									if( 0 == strncasecmp(device_id, msg->u.crud.dest, device_id_len-1) )
									{
										 /* For this device. */
										parodus_len = strlen( PARODUS_SERVICE_NAME );
										if( 0 == strncmp(PARODUS_SERVICE_NAME, &msg->u.crud.dest[device_id_len], parodus_len-1) ) 
										{
											/* For Parodus CRUD queue. */
											ParodusInfo("Create RetrieveMsg and add to parodus CRUD queue\n");
											createUpstreamRetrieveMsg(msg, &retrieve_msg);
											addCRUDmsgToQueue(retrieve_msg);
										}
										else
										{
											/* For nanomsg clients. */
											getServiceNameAndSendResponse(msg, &message->msg, message->len);
										}
									}
									else
									{
										/* Not for this device. Send upstream */
										ParodusInfo("sendUpstreamMsgToServer \n");
										sendUpstreamMsgToServer(&message->msg, message->len);
									}
									if(device_id != NULL)
									{
										free(device_id);
										device_id = NULL;
									}
								}
								else
								{
									ParodusError("Failed to get device_id\n");
								}
						}
						else if((WRP_MSG_TYPE__UPDATE == msgType) || (WRP_MSG_TYPE__DELETE == msgType))
						{
							ParodusPrint("UPDATE/DELETE request\n");
							ret = getDeviceId(&device_id, &device_id_len);
							if(ret == 0)
							{
								ParodusPrint("device_id %s device_id_len %lu\n", device_id, device_id_len);
								/* Match dest based on device_id. Check dest start with: "mac:112233445xxx" ? */
								if( 0 == strncasecmp(device_id, msg->u.crud.dest, device_id_len-1) )
								{
									/* For this device. For nanomsg clients.*/
									getServiceNameAndSendResponse(msg, &message->msg, message->len);
								}
								else
								{
									/* Not for this device. Send upstream */
									ParodusInfo("sendUpstreamMsgToServer \n");
									sendUpstreamMsgToServer(&message->msg, message->len);
								}
								if(device_id != NULL)
								{
									free(device_id);
									device_id = NULL;
								}
							}
							else
							{
								ParodusError("Failed to get device_id\n");
							}
						 }
						 else if (WRP_MSG_TYPE__SVC_ALIVE != msgType) {
						  /* Don't reply to service alive message */
							sendUpstreamMsgToServer(&message->msg, message->len);
					}
			  }
		    }
           	}
            else
            {
                ParodusError("Error in msgpack decoding for upstream\n");
            }

			//nn_freemsg should not be done for parodus/tags/ CRUD requests as it is not received through nanomsg.
			if ((msg && (msg->u.crud.source !=NULL) && wrp_does_service_match("parodus", msg->u.crud.source) == 0))
			{
				free(message->msg);
			}
			else
			{
				if(nn_freemsg (message->msg) < 0)
				{
					ParodusError ("Failed to free msg\n");
				}
			}
			ParodusPrint("Free for upstream decoded msg\n");
			if (msg) {
                wrp_free_struct(msg);
            }
			msg = NULL;
			free(message);
			message = NULL;
        }
        else
        {
            if (g_shutdown) {
                pthread_mutex_unlock (&nano_mut);
                break;
            }
            ParodusPrint("Before pthread cond wait in consumer thread\n");   
            pthread_cond_wait(&nano_con, &nano_mut);
            pthread_mutex_unlock (&nano_mut);
            ParodusPrint("mutex unlock in consumer thread after cond wait\n");
        }
    }
    return NULL;
}

/**
 * @brief getDeviceId function to create deviceId in the format "mac:112233445xxx"
 *
 * @param[out] device_id in the format "mac:112233445xxx"
 * @param[out] total size of device_id
 */
int getDeviceId(char **device_id, size_t *device_id_len)
{
	char *deviceID = NULL;
	size_t len;

	if((get_parodus_cfg()->hw_mac !=NULL) && (strlen(get_parodus_cfg()->hw_mac)!=0))
	{
		len = strlen(get_parodus_cfg()->hw_mac) + 5;

		deviceID = (char *) malloc(sizeof(char)*64);
		if(deviceID != NULL)
		{
			snprintf(deviceID, len, "mac:%s", get_parodus_cfg()->hw_mac);
			*device_id = deviceID;
			*device_id_len = len;
		}
		else
		{
			ParodusError("device_id allocation failed\n");
			return -1;
		}
	}
	else
	{
		ParodusError("device mac is NULL\n");
		return -1;
	}
	return 0;
}


/**
 * @brief createUpstreamRetrieveMsg function to create new message for processing Retrieve requests
 *
 * @param[in] message The upstream message received from cloud or internal clients
 * @param[out] retrieve_msg New message for processing Retrieve requests
 */
void createUpstreamRetrieveMsg(wrp_msg_t *message, wrp_msg_t **retrieve_msg)
{
	wrp_msg_t *msg;
	msg = ( wrp_msg_t * ) malloc( sizeof( wrp_msg_t ) );
	if(msg != NULL)
    {
		memset(msg, 0, sizeof(wrp_msg_t));

		msg->msg_type = message->msg_type;
		if(message->u.crud.transaction_uuid != NULL)
		{
			msg->u.crud.transaction_uuid = strdup(message->u.crud.transaction_uuid);
		}
		if(message->u.crud.source !=NULL)
		{
			msg->u.crud.source = strdup(message->u.crud.source);
		}
		if(message->u.crud.dest != NULL)
		{
			msg->u.crud.dest = strdup(message->u.crud.dest);
		}
		*retrieve_msg = msg;
	}
}

/**
 * @brief getServiceNameAndSendResponse function to fetch client service name and to send msg to it.
 *
 * @param[in] msg The decoded message to fetch client service name from its dest field
 * @param[in] msg_bytes The encoded upstream msg to be sent to client
 * @param[in] msg_size Total size of the msg to send to client
 */
void getServiceNameAndSendResponse(wrp_msg_t *msg, void **msg_bytes, size_t msg_size)
{
	char *serviceName = NULL;
	int sendStatus =-1;

	serviceName = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, DEST);
	if ( serviceName != NULL)
	{
		sendStatus=sendMsgtoRegisteredClients(serviceName,(const char **)msg_bytes, msg_size);
		if(sendStatus ==1)
		{
			ParodusInfo("Send upstreamMsg successfully to registered client %s\n", serviceName);
		}
		else
		{
			ParodusError("Failed to send upstreamMsg to registered client %s\n", serviceName);
		}
		free(serviceName);
		serviceName = NULL;
	}
	else
	{
		ParodusError("serviceName is NULL,not sending retrieve response to client\n");
	}
}

int sendUpstreamMsgToServer(void **resp_bytes, size_t resp_size)
{
	void *appendData;
	size_t encodedSize;
	bool close_retry = false;
	int sendRetStatus = 1;
	//appending response with metadata 			
	if(metaPackSize > 0)
	{
	   	encodedSize = appendEncodedData( &appendData, *resp_bytes, resp_size, metadataPack, metaPackSize );
	   	ParodusPrint("metadata appended upstream response %s\n", (char *)appendData);
	   	ParodusPrint("encodedSize after appending :%zu\n", encodedSize);
	   		   
		ParodusInfo("Sending response to server\n");
		close_retry = get_close_retry();

		/* send response when connection retry is not in progress. Also during cloud_disconnect UPDATE request. Here, close_retry becomes 1 hence check is added to send disconnect response to server. */
		//TODO: Upstream and downstream messages in queue should be handled and queue should be empty before parodus forcefully disconnect from cloud.
		if(!close_retry || (get_parodus_cfg()->cloud_disconnect !=NULL))
		{
			sendRetStatus = sendMessage(get_global_conn(),appendData, encodedSize);
		}
		else
		{
			ParodusInfo("close_retry is %d, unable to send response as connection retry is in progress\n", close_retry);
			OnboardLog("close_retry is %d, unable to send response as connection retry is in progress\n", close_retry);
			sendRetStatus = 1;
		}
		free(appendData);
		appendData =NULL;
	}
	else
	{		
		ParodusError("Failed to send upstream as metadata packing is not successful\n");
		sendRetStatus = 1;
	}
	ParodusPrint("sendRetStatus is %d\n", sendRetStatus);

	return sendRetStatus;
}