enhance(header): Add message type

[ric-plt/lib/rmr.git] / src / rmr / nng / src / rmr_nng.c

// : vi ts=4 sw=4 noet :
/*
==================================================================================
        Copyright (c) 2019 Nokia
        Copyright (c) 2018-2019 AT&T Intellectual Property.

   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.
==================================================================================
*/

/*
        Mnemonic:       rmr_nng.c
        Abstract:       This is the compile point for the nng version of the rmr
                                library (formarly known as uta, so internal function names
                                are likely still uta_*)

                                With the exception of the symtab portion of the library,
                                RMr is built with a single compile so as to "hide" the
                                internal functions as statics.  Because they interdepend
                                on each other, and CMake has issues with generating two
                                different wormhole objects from a single source, we just
                                pull it all together with a centralised comple using
                                includes.

                                Future:  the API functions at this point can be separated
                                into a common source module.

        Author:         E. Scott Daniels
        Date:           1 February 2019
*/

#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <netdb.h>
#include <errno.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <unistd.h>
#include <time.h>
#include <arpa/inet.h>

#include <nng/nng.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <nng/protocol/pipeline0/push.h>
#include <nng/protocol/pipeline0/pull.h>


#include "rmr.h"                                // things the users see
#include "rmr_agnostic.h"               // agnostic things (must be included before private)
#include "rmr_nng_private.h"    // things that we need too
#include "rmr_symtab.h"

#include "ring_static.c"                        // message ring support
#include "rt_generic_static.c"          // route table things not transport specific
#include "rtable_nng_static.c"          // route table things -- transport specific
#include "rtc_static.c"                         // route table collector
#include "tools_static.c"
#include "sr_nng_static.c"                      // send/receive static functions
#include "wormholes.c"                          // wormhole api externals and related static functions (must be LAST!)


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


/*
        Clean up a context.
*/
static void free_ctx( uta_ctx_t* ctx ) {
        if( ctx ) {
                if( ctx->rtg_addr ) {
                        free( ctx->rtg_addr );
                }
        }
}

// --------------- public functions --------------------------------------------------------------------------

/*
        Returns the size of the payload (bytes) that the msg buffer references.
        Len in a message is the number of bytes which were received, or should
        be transmitted, however, it is possible that the mbuf was allocated
        with a larger payload space than the payload length indicates; this
        function returns the absolute maximum space that the user has available
        in the payload. On error (bad msg buffer) -1 is returned and errno should
        indicate the rason.
*/
extern int rmr_payload_size( rmr_mbuf_t* msg ) {
        if( msg == NULL || msg->header == NULL ) {
                errno = EINVAL;
                return -1;
        }

        errno = 0;
        return msg->alloc_len - RMR_HDR_LEN( msg->header );                             // allocated transport size less the header and other data bits
}

/*
        Allocates a send message as a zerocopy message allowing the underlying message protocol
        to send the buffer without copy.
*/
extern rmr_mbuf_t* rmr_alloc_msg( void* vctx, int size ) {
        uta_ctx_t*      ctx;
        rmr_mbuf_t*     m;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return NULL;
        }

        m = alloc_zcmsg( ctx, NULL, size, 0, DEF_TR_LEN );                              // alloc with default trace data
        return  m;
}


/*
        Allocates a send message as a zerocopy message allowing the underlying message protocol
        to send the buffer without copy. In addition, a trace data field of tr_size will be
        added and the supplied data coppied to the buffer before returning the message to
        the caller.
*/
extern rmr_mbuf_t* rmr_tralloc_msg( void* vctx, int size, int tr_size, unsigned const char* data ) {
        uta_ctx_t*      ctx;
        rmr_mbuf_t*     m;
        int state;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return NULL;
        }

        m = alloc_zcmsg( ctx, NULL, size, 0, tr_size );                         // alloc with specific tr size
        if( m != NULL ) {
                state = rmr_set_trace( m, data, tr_size );                              // roll their data in
                if( state != tr_size ) {
                        m->state = RMR_ERR_INITFAILED;
                }
        }

        return  m;
}

/*
        This provides an external path to the realloc static function as it's called by an
        outward facing mbuf api function. Used to reallocate a message with a different
        trace data size.
*/
extern rmr_mbuf_t* rmr_realloc_msg( rmr_mbuf_t* msg, int new_tr_size ) {
        return realloc_msg( msg, new_tr_size );
}


/*
        Return the message to the available pool, or free it outright.
*/
extern void rmr_free_msg( rmr_mbuf_t* mbuf ) {
        if( mbuf == NULL ) {
                return;
        }

        if( mbuf->header ) {
                if( mbuf->flags & MFL_ZEROCOPY ) {
                        //nng_free( (void *) mbuf->header, mbuf->alloc_len );
                        if( mbuf->tp_buf ) {
                                nng_msg_free(  mbuf->tp_buf );
                        }
                }
        }

        free( mbuf );
}

/*
        send message with maximum timeout.
        Accept a message and send it to an endpoint based on message type.
        If NNG reports that the send attempt timed out, or should be retried,
        RMr will retry for approximately max_to microseconds; rounded to the next
        higher value of 10.

        Allocates a new message buffer for the next send. If a message type has
        more than one group of endpoints defined, then the message will be sent
        in round robin fashion to one endpoint in each group.

        An endpoint will be looked up in the route table using the message type and
        the subscription id. If the subscription id is "UNSET_SUBID", then only the
        message type is used.  If the initial lookup, with a subid, fails, then a
        second lookup using just the mtype is tried.

        CAUTION: this is a non-blocking send.  If the message cannot be sent, then
                it will return with an error and errno set to eagain. If the send is
                a limited fanout, then the returned status is the status of the last
                send attempt.

*/
extern rmr_mbuf_t* rmr_mtosend_msg( void* vctx, rmr_mbuf_t* msg, int max_to ) {
        nng_socket nn_sock;                     // endpoint socket for send
        uta_ctx_t*      ctx;
        int     group;                                  // selected group to get socket for
        int send_again;                         // true if the message must be sent again
        rmr_mbuf_t*     clone_m;                // cloned message for an nth send
        int sock_ok;                            // got a valid socket from round robin select
        uint64_t key;                           // mtype or sub-id/mtype sym table key
        int     altk_ok = 0;                    // set true if we can lookup on alternate key if mt/sid lookup fails

        if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
                errno = EINVAL;                                                                                         // if msg is null, this is their clue
                if( msg != NULL ) {
                        msg->state = RMR_ERR_BADARG;
                        errno = EINVAL;                                                                                 // must ensure it's not eagain
                }
                return msg;
        }

        errno = 0;                                                                                                      // clear; nano might set, but ensure it's not left over if it doesn't
        if( msg->header == NULL ) {
                fprintf( stderr, "rmr_send_msg: ERROR: message had no header\n" );
                msg->state = RMR_ERR_NOHDR;
                errno = EBADMSG;                                                                                        // must ensure it's not eagain
                return msg;
        }

        if( max_to < 0 ) {
                max_to = ctx->send_retries;             // convert to retries
        }

        send_again = 1;                                                                                 // force loop entry
        group = 0;                                                                                              // always start with group 0

        key = build_rt_key( msg->sub_id, msg->mtype );                  // route table key to find the entry
        if( msg->sub_id != UNSET_SUBID ) {
                altk_ok = 1;                                                                            // if caller's sub-id doesn't hit with mtype, allow mtype only key for retry
        }
        while( send_again ) {
                sock_ok = uta_epsock_rr( ctx->rtable, key, group, &send_again, &nn_sock );              // round robin sel epoint; again set if mult groups
                if( DEBUG ) fprintf( stderr, "[DBUG] send msg: type=%d again=%d group=%d len=%d sock_ok=%d ak_ok=%d\n",
                                msg->mtype, send_again, group, msg->len, sock_ok, altk_ok );

                if( ! sock_ok ) {
                        if( altk_ok ) {                                                                                 // we can try with the alternate (no sub-id) key
                                altk_ok = 0;
                                key = build_rt_key( UNSET_SUBID, msg->mtype );          // build with just the mtype and try again
                                send_again = 1;                                                                         // ensure we don't exit the while
                                continue;
                        }

                        msg->state = RMR_ERR_NOENDPT;
                        errno = ENXIO;                                                                                  // must ensure it's not eagain
                        return msg;                                                                                             // caller can resend (maybe) or free
                }

                group++;

                if( send_again ) {
                        clone_m = clone_msg( msg );                                                             // must make a copy as once we send this message is not available
                        if( DEBUG ) fprintf( stderr, "[DBUG] msg cloned: type=%d len=%d\n", msg->mtype, msg->len );
                        msg->flags |= MFL_NOALLOC;                                                              // send should not allocate a new buffer
                        msg = send_msg( ctx, msg, nn_sock, max_to );                    // do the hard work, msg should be nil on success
                        /*
                        if( msg ) {
                                // error do we need to count successes/errors, how to report some success, esp if last fails?
                        }
                        */

                        msg = clone_m;                                                                                  // clone will be the next to send
                } else {
                        msg = send_msg( ctx, msg, nn_sock, max_to );                    // send the last, and allocate a new buffer; drops the clone if it was
                }
        }

        return msg;                                                                     // last message caries the status of last/only send attempt
}

/*
        Send with default max timeout as is set in the context.
        See rmr_mtosend_msg() for more details on the parameters.
        See rmr_stimeout() for info on setting the default timeout.
*/
extern rmr_mbuf_t* rmr_send_msg( void* vctx, rmr_mbuf_t* msg ) {
        return rmr_mtosend_msg( vctx, msg,  -1 );                       // retries <  uses default from ctx
}

/*
        Return to sender allows a message to be sent back to the endpoint where it originated.
        The source information in the message is used to select the socket on which to write
        the message rather than using the message type and round-robin selection. This
        should return a message buffer with the state of the send operation set. On success
        (state is RMR_OK, the caller may use the buffer for another receive operation), and on
        error it can be passed back to this function to retry the send if desired. On error,
        errno will liklely have the failure reason set by the nng send processing.
        The following are possible values for the state in the message buffer:

        Message states returned:
                RMR_ERR_BADARG - argument (context or msg) was nil or invalid
                RMR_ERR_NOHDR  - message did not have a header
                RMR_ERR_NOENDPT- an endpoint to send the message to could not be determined
                RMR_ERR_SENDFAILED - send failed; errno has nano error code
                RMR_ERR_RETRY   - the reqest failed but should be retried (EAGAIN)

        A nil message as the return value is rare, and generally indicates some kind of horrible
        failure. The value of errno might give a clue as to what is wrong.

        CAUTION:
                Like send_msg(), this is non-blocking and will return the msg if there is an errror.
                The caller must check for this and handle.
*/
extern rmr_mbuf_t*  rmr_rts_msg( void* vctx, rmr_mbuf_t* msg ) {
        nng_socket nn_sock;                     // endpoint socket for send
        uta_ctx_t*      ctx;
        int state;
        uta_mhdr_t*     hdr;
        char*   hold_src;                       // we need the original source if send fails
        int             sock_ok;                        // true if we found a valid endpoint socket

        if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
                errno = EINVAL;                                                                                         // if msg is null, this is their clue
                if( msg != NULL ) {
                        msg->state = RMR_ERR_BADARG;
                }
                return msg;
        }

        errno = 0;                                                                                                              // at this point any bad state is in msg returned
        if( msg->header == NULL ) {
                fprintf( stderr, "[ERR] rmr_send_msg: message had no header\n" );
                msg->state = RMR_ERR_NOHDR;
                return msg;
        }

        sock_ok = uta_epsock_byname( ctx->rtable, (char *) ((uta_mhdr_t *)msg->header)->src, &nn_sock );                        // socket of specific endpoint
        if( ! sock_ok ) {
                msg->state = RMR_ERR_NOENDPT;
                return msg;                                                     // preallocated msg can be reused since not given back to nn
        }

        msg->state = RMR_OK;                                    // ensure it is clear before send
        hold_src = strdup( (char *) ((uta_mhdr_t *)msg->header)->src );                                                 // the dest where we're returning the message to
        strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SID );                        // must overlay the source to be ours
        msg = send_msg( ctx, msg, nn_sock, -1 );
        if( msg ) {
                strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, hold_src, RMR_MAX_SID );                    // always return original source so rts can be called again
                msg->flags |= MFL_ADDSRC;                                                                                                       // if msg given to send() it must add source
        }

        free( hold_src );
        return msg;
}

/*
        Call sends the message based on message routing using the message type, and waits for a
        response message to arrive with the same transaction id that was in the outgoing message.
        If, while wiating for the expected response,  messages are received which do not have the
        desired transaction ID, they are queued. Calls to uta_rcv_msg() will dequeue them in the
        order that they were received.

        Normally, a message struct pointer is returned and msg->state must be checked for RMR_OK
        to ensure that no error was encountered. If the state is UTA_BADARG, then the message
        may be resent (likely the context pointer was nil).  If the message is sent, but no
        response is received, a nil message is returned with errno set to indicate the likley
        issue:
                ETIMEDOUT -- too many messages were queued before reciving the expected response
                ENOBUFS -- the queued message ring is full, messages were dropped
                EINVAL  -- A parameter was not valid
                EAGAIN  -- the underlying message system wsa interrupted or the device was busy;
                                        user should call this function with the message again.


        QUESTION:  should user specify the number of messages to allow to queue?
*/
extern rmr_mbuf_t* rmr_call( void* vctx, rmr_mbuf_t* msg ) {
        uta_ctx_t*              ctx;
        unsigned char   expected_id[RMR_MAX_XID+1];             // the transaction id in the message; we wait for response with same ID

        if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
                if( msg != NULL ) {
                        msg->state = RMR_ERR_BADARG;
                }
                return msg;
        }

        memcpy( expected_id, msg->xaction, RMR_MAX_XID );
        expected_id[RMR_MAX_XID] = 0;                                   // ensure it's a string
        if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rmr_call is making call, waiting for (%s)\n", expected_id );
        errno = 0;
        msg->flags |= MFL_NOALLOC;                                              // we don't need a new buffer from send

        msg = rmr_send_msg( ctx, msg );
        if( msg ) {                                                                             // msg should be nil, if not there was a problem; return buffer to user
                if( msg->state != RMR_ERR_RETRY ) {
                        msg->state = RMR_ERR_CALLFAILED;                // errno not available to all wrappers; don't stomp if marked retry
                }
                return msg;
        }

        return rmr_rcv_specific( ctx, NULL, (char *) expected_id, 20 );                 // wait for msg allowing 20 to queue ahead
}

/*
        The outward facing receive function. When invoked it will pop the oldest message
        from the receive ring, if any are queued, and return it. If the ring is empty
        then the receive function is invoked to wait for the next message to arrive (blocking).

        If old_msg is provided, it will be populated (avoiding lots of free/alloc cycles). If
        nil, a new one will be allocated. However, the caller should NOT expect to get the same
        struct back (if a queued message is returned the message struct will be different).
*/
extern rmr_mbuf_t* rmr_rcv_msg( void* vctx, rmr_mbuf_t* old_msg ) {
        uta_ctx_t*      ctx;
        rmr_mbuf_t*     qm;                             // message that was queued on the ring

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                if( old_msg != NULL ) {
                        old_msg->state = RMR_ERR_BADARG;
                }
                errno = EINVAL;
                return old_msg;
        }
        errno = 0;

        qm = (rmr_mbuf_t *) uta_ring_extract( ctx->mring );                     // pop if queued
        if( qm != NULL ) {
                if( old_msg ) {
                        rmr_free_msg( old_msg );                                                        // future:  push onto a free list???
                }

                return qm;
        }

        return rcv_msg( ctx, old_msg );                                                         // nothing queued, wait for one
}

/*
        This implements a receive with a timeout via epoll. Mostly this is for
        wrappers as native C applications can use epoll directly and will not have
        to depend on this.
*/
extern rmr_mbuf_t* rmr_torcv_msg( void* vctx, rmr_mbuf_t* old_msg, int ms_to ) {
        struct epoll_stuff* eps;        // convience pointer
        uta_ctx_t*      ctx;
        rmr_mbuf_t*     qm;                             // message that was queued on the ring
        int nready;
        rmr_mbuf_t* msg;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                if( old_msg != NULL ) {
                        old_msg->state = RMR_ERR_BADARG;
                }
                errno = EINVAL;
                return old_msg;
        }

        qm = (rmr_mbuf_t *) uta_ring_extract( ctx->mring );                     // pop if queued
        if( qm != NULL ) {
                if( old_msg ) {
                        rmr_free_msg( old_msg );                                                        // future:  push onto a free list???
                }

                return qm;
        }

        if( (eps = ctx->eps)  == NULL ) {                                       // set up epoll on first call
                eps = malloc( sizeof *eps );

                if( (eps->ep_fd = epoll_create1( 0 )) < 0 ) {
                fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
                        free( eps );
                        return NULL;
                }

                eps->nng_fd = rmr_get_rcvfd( ctx );
                eps->epe.events = EPOLLIN;
                eps->epe.data.fd = eps->nng_fd;

                if( epoll_ctl( eps->ep_fd, EPOLL_CTL_ADD, eps->nng_fd, &eps->epe ) != 0 )  {
                fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
                        free( eps );
                        return NULL;
                }

                ctx->eps = eps;
        }

        if( old_msg ) {
                msg = old_msg;
        } else {
                msg = alloc_zcmsg( ctx, NULL, RMR_MAX_RCV_BYTES, RMR_OK, DEF_TR_LEN );                  // will abort on failure, no need to check
        }

        if( ms_to < 0 ) {
                ms_to = 0;
        }

        nready = epoll_wait( eps->ep_fd, eps->events, 1, ms_to );     // block until something or timedout
        if( nready <= 0 ) {                                             // we only wait on ours, so we assume ready means it's ours
                msg->state = RMR_ERR_TIMEOUT;
        } else {
                return rcv_msg( ctx, msg );                                                             // receive it and return it
        }

        return msg;                             // return empty message with state set
}

/*
        This blocks until the message with the 'expect' ID is received. Messages which are received
        before the expected message are queued onto the message ring.  The function will return
        a nil message and set errno to ETIMEDOUT if allow2queue messages are received before the
        expected message is received. If the queued message ring fills a nil pointer is returned
        and errno is set to ENOBUFS.

        Generally this will be invoked only by the call() function as it waits for a response, but
        it is exposed to the user application as three is no reason not to.
*/
extern rmr_mbuf_t* rmr_rcv_specific( void* vctx, rmr_mbuf_t* msg, char* expect, int allow2queue ) {
        uta_ctx_t*      ctx;
        int     queued = 0;                             // number we pushed into the ring
        int     exp_len = 0;                    // length of expected ID

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                if( msg != NULL ) {
                        msg->state = RMR_ERR_BADARG;
                }
                errno = EINVAL;
                return msg;
        }

        errno = 0;

        if( expect == NULL || ! *expect ) {                             // nothing expected if nil or empty string, just receive
                return rmr_rcv_msg( ctx, msg );
        }

        exp_len = strlen( expect );
        if( exp_len > RMR_MAX_XID ) {
                exp_len = RMR_MAX_XID;
        }
        if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific waiting for id=%s\n",  expect );

        while( queued < allow2queue ) {
                msg = rcv_msg( ctx, msg );                                      // hard wait for next
                if( msg->state == RMR_OK ) {
                        if( memcmp( msg->xaction, expect, exp_len ) == 0 ) {                    // got it -- return it
                                if( DEBUG ) fprintf( stderr, "[DBUG] rcv-specific matched (%s); %d messages were queued\n", msg->xaction, queued );
                                return msg;
                        }

                        if( ! uta_ring_insert( ctx->mring, msg ) ) {                                    // just queue, error if ring is full
                                if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_specific ring is full\n" );
                                errno = ENOBUFS;
                                return NULL;
                        }

                        if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific queued message type=%d\n", msg->mtype );
                        queued++;
                        msg = NULL;
                }
        }

        if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific timeout waiting for %s\n", expect );
        errno = ETIMEDOUT;
        return NULL;
}

//  CAUTION:  these are not supported as they must be set differently (between create and open) in NNG.
//                              until those details are worked out, these generate a warning.
/*
        Set send timeout. The value time is assumed to be microseconds.  The timeout is the
        rough maximum amount of time that RMr will block on a send attempt when the underlying
        mechnism indicates eagain or etimeedout.  All other error conditions are reported
        without this delay. Setting a timeout of 0 causes no retries to be attempted in
        RMr code. Setting a timeout of 1 causes RMr to spin up to 10K retries before returning,
        but without issuing a sleep.  If timeout is > 1, then RMr will issue a sleep (1us)
        after every 10K send attempts until the time value is reached. Retries are abandoned
        if NNG returns anything other than NNG_AGAIN or NNG_TIMEDOUT.

        The default, if this function is not used, is 1; meaning that RMr will retry, but will
        not enter a sleep.  In all cases the caller should check the status in the message returned
        after a send call.

        Returns -1 if the context was invalid; RMR_OK otherwise.
*/
extern int rmr_set_stimeout( void* vctx, int time ) {
        uta_ctx_t*      ctx;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return -1;
        }

        if( time < 0 ) {
                time = 0;
        }

        ctx->send_retries = time;
        return RMR_OK;
}

/*
        Set receive timeout -- not supported in nng implementation
*/
extern int rmr_set_rtimeout( void* vctx, int time ) {
        fprintf( stderr, "[WRN] Current implementation of RMR ontop of NNG does not support setting a receive timeout\n" );
        return 0;
}


/*
        This is the actual init workhorse. The user visible function meerly ensures that the
        calling programme does NOT set any internal flags that are supported, and then
        invokes this.  Internal functions (the route table collector) which need additional
        open ports without starting additional route table collectors, will invoke this
        directly with the proper flag.
*/
static void* init(  char* uproto_port, int max_msg_size, int flags ) {
        static  int announced = 0;
        uta_ctx_t*      ctx = NULL;
        char    bind_info[NNG_MAXADDRLEN];      // bind info
        char*   proto = "tcp";                          // pointer into the proto/port string user supplied
        char*   port;
        char*   interface = NULL;                       // interface to bind to (from RMR_BIND_IF, 0.0.0.0 if not defined)
        char*   proto_port;
        char    wbuf[1024];                                     // work buffer
        char*   tok;                                            // pointer at token in a buffer
        int             state;

        if( ! announced ) {
                fprintf( stderr, "[INFO] ric message routing library on NNG mv=%d (%s %s.%s.%s built: %s)\n",
                        RMR_MSG_VER, QUOTE_DEF(GIT_ID), QUOTE_DEF(MAJOR_VER), QUOTE_DEF(MINOR_VER), QUOTE_DEF(PATCH_VER), __DATE__ );
                announced = 1;
        }

        errno = 0;
        if( uproto_port == NULL ) {
                proto_port = strdup( DEF_COMM_PORT );
        } else {
                proto_port = strdup( uproto_port );             // so we can modify it
        }

        if( (ctx = (uta_ctx_t *) malloc( sizeof( uta_ctx_t ) )) == NULL ) {
                errno = ENOMEM;
                return NULL;
        }
        memset( ctx, 0, sizeof( uta_ctx_t ) );

        ctx->send_retries = 1;                                                  // default is not to sleep at all; RMr will retry about 10K times before returning
        ctx->mring = uta_mk_ring( 128 );                                // message ring to hold asynch msgs received while waiting for call response

        ctx->max_plen = RMR_MAX_RCV_BYTES;                              // max user payload lengh
        if( max_msg_size > 0 ) {
                ctx->max_plen = max_msg_size;
        }

        // we're using a listener to get rtg updates, so we do NOT need this.
        //uta_lookup_rtg( ctx );                                                        // attempt to fill in rtg info; rtc will handle missing values/errors

        if( nng_pull0_open( &ctx->nn_sock )  !=  0 ) {          // and assign the mode
                fprintf( stderr, "[CRI] rmr_init: unable to initialise nng listen (pull) socket: %d\n", errno );
                free_ctx( ctx );
                return NULL;
        }

        if( (port = strchr( proto_port, ':' )) != NULL ) {
                if( port == proto_port ) {              // ":1234" supplied; leave proto to default and point port correctly
                        port++;
                } else {
                        *(port++) = 0;                  // term proto string and point at port string
                        proto = proto_port;             // user supplied proto so point at it rather than default
                }
        } else {
                port = proto_port;                      // assume something like "1234" was passed
        }

        if( (gethostname( wbuf, sizeof( wbuf ) )) != 0 ) {
                fprintf( stderr, "[CRI] rmr_init: cannot determine localhost name: %s\n", strerror( errno ) );
                return NULL;
        }
        if( (tok = strchr( wbuf, '.' )) != NULL ) {
                *tok = 0;                                                                       // we don't keep domain portion
        }
        ctx->my_name = (char *) malloc( sizeof( char ) * RMR_MAX_SID );
        if( snprintf( ctx->my_name, RMR_MAX_SID, "%s:%s", wbuf, port ) >= RMR_MAX_SID ) {                       // our registered name is host:port
                fprintf( stderr, "[CRI] rmr_init: hostname + port must be less than %d characters; %s:%s is not\n", RMR_MAX_SID, wbuf, port );
                return NULL;
        }

        ctx->ip_list = mk_ip_list( port );              // suss out all IP addresses we can find on the box, and bang on our port for RT comparisons



        if( (interface = getenv( ENV_BIND_IF )) == NULL ) {
                interface = "0.0.0.0";
        }
        // NOTE: if there are options that might need to be configured, the listener must be created, options set, then started
        //       rather than using this generic listen() call.
        snprintf( bind_info, sizeof( bind_info ), "%s://%s:%s", proto, interface, port );
        if( (state = nng_listen( ctx->nn_sock, bind_info, NULL, NO_FLAGS )) != 0 ) {
                fprintf( stderr, "[CRIT] rmr_init: unable to start nng listener for %s: %s\n", bind_info, nng_strerror( state ) );
                nng_close( ctx->nn_sock );
                free_ctx( ctx );
                return NULL;
        }

        if( !(flags & FL_NOTHREAD) ) {                                                                          // skip if internal function that doesnt need an rtc
                if( pthread_create( &ctx->rtc_th,  NULL, rtc, (void *) ctx ) ) {        // kick the rt collector thread
                        fprintf( stderr, "[WARN] rmr_init: unable to start route table collector thread: %s", strerror( errno ) );
                }
        }

        free( proto_port );
        return (void *) ctx;
}

/*
        Initialise the message routing environment. Flags are one of the UTAFL_
        constants. Proto_port is a protocol:port string (e.g. tcp:1234). If default protocol
        (tcp) to be used, then :port is all that is needed.

        At the moment it seems that TCP really is the only viable protocol, but
        we'll allow flexibility.

        The return value is a void pointer which must be passed to most uta functions. On
        error, a nil pointer is returned and errno should be set.

        Flags:
                No user flags supported (needed) at the moment, but this provides for extension
                without drastically changing anything. The user should invoke with RMRFL_NONE to
                avoid any misbehavour as there are internal flags which are suported
*/
extern void* rmr_init( char* uproto_port, int max_msg_size, int flags ) {
        return init( uproto_port, max_msg_size, flags & UFL_MASK  );            // ensure any internal flags are off
}

/*
        This sets the default trace length which will be added to any message buffers
        allocated.  It can be set at any time, and if rmr_set_trace() is given a
        trace len that is different than the default allcoated in a message, the message
        will be resized.

        Returns 0 on failure and 1 on success. If failure, then errno will be set.
*/
extern int rmr_init_trace( void* vctx, int tr_len ) {
        uta_ctx_t* ctx;

        errno = 0;
        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                errno = EINVAL;
                return 0;
        }

        ctx->trace_data_len = tr_len;
        return 1;
}

/*
        Return true if routing table is initialised etc. and app can send/receive.
*/
extern int rmr_ready( void* vctx ) {
        uta_ctx_t *ctx;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return FALSE;
        }

        if( ctx->rtable != NULL ) {
                return TRUE;
        }

        return FALSE;
}

/*
        Returns a file descriptor which can be used with epoll() to signal a receive
        pending. The file descriptor should NOT be read from directly, nor closed, as NNG
        does not support this.
*/
extern int rmr_get_rcvfd( void* vctx ) {
        uta_ctx_t* ctx;
        int fd;
        int state;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return -1;
        }

        if( (state = nng_getopt_int( ctx->nn_sock, NNG_OPT_RECVFD, &fd )) != 0 ) {
                fprintf( stderr, "[WRN] rmr cannot get recv fd: %s\n", nng_strerror( state ) );
                return -1;
        }

        return fd;
}


/*
        Clean up things.

        There isn't an nng_flush() per se, but we can pause, generate
        a context switch, which should allow the last sent buffer to
        flow. There isn't exactly an nng_term/close either, so there
        isn't much we can do.
*/
extern void rmr_close( void* vctx ) {
        uta_ctx_t *ctx;

        if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
                return;
        }

        ctx->shutdown = 1;
        nng_close( ctx->nn_sock );
}