src/upstream.c
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;
}
/*
 * @brief To handle UpStream messages which is received from nanomsg server socket
 */
@brief To handle UpStream messages which is received from nanomsg server socket
@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);
                    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 * handle_upstream ( ) { UpStreamMsg * message ; int sock , bind ; int bytes = 0 ; void * buf ; ParodusPrint ( " \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 ( " \n " , errno , strerror ( errno ) ) ; } ParodusPrint ( " \n " , get_parodus_cfg ( ) -> local_url ) ; bind = nn_bind ( sock , get_parodus_cfg ( ) -> local_url ) ; if ( bind < 0 ) { ParodusError ( " \n " , errno , strerror ( errno ) ) ; } else { ParodusInfo ( " \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 ( " \n " , errno ) ; continue ; } ParodusInfo ( " \n " ) ; message = ( UpStreamMsg * ) malloc ( sizeof ( UpStreamMsg ) ) ; if ( message ) { message -> msg = buf ; message -> len = bytes ; message -> next = NULL ; pthread_mutex_lock ( & nano_mut ) ; if ( UpStreamMsgQ == NULL ) { UpStreamMsgQ = message ; ParodusPrint ( " \n " ) ; pthread_cond_signal ( & nano_con ) ; pthread_mutex_unlock ( & nano_mut ) ; ParodusPrint ( " \n " ) ; } else { UpStreamMsg * temp = UpStreamMsgQ ; while ( temp -> next ) { temp = temp -> next ; } temp -> next = message ; pthread_mutex_unlock ( & nano_mut ) ; } } else { ParodusError ( " \n " ) ; } } if ( nn_shutdown ( sock , bind ) < 0 ) { ParodusError ( " \n " , sock , bind , errno ) ; } if ( nn_close ( sock ) < 0 ) { ParodusError ( " \n " , sock , errno ) ; } } } else { ParodusError ( " \n " , errno , strerror ( errno ) ) ; } ParodusPrint ( " \n " ) ; return 0 ; }
@brief To handle UpStream messages which is received from nanomsg server socket
@brief To handle UpStream messages which is received from nanomsg server socket
//Producer adds the nanoMsg into queue