src/nanomsg/src/rmr.c

   1 // :vi sw=4 ts=4 noet:
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2019 Nokia
   5         Copyright (c) 2018-2019 AT&T Intellectual Property.
   6
   7    Licensed under the Apache License, Version 2.0 (the "License");
   8    you may not use this file except in compliance with the License.
   9    You may obtain a copy of the License at
  10
  11            http://www.apache.org/licenses/LICENSE-2.0
  12
  13    Unless required by applicable law or agreed to in writing, software
  14    distributed under the License is distributed on an "AS IS" BASIS,
  15    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16    See the License for the specific language governing permissions and
  17    limitations under the License.
  18 ==================================================================================
  19 */
  20
  21 /*
  22         Mnemonic:       rmr.c
  23         Abstract:       The bulk of the ric message routing library which is built upon
  24                                 the older nanomsg messaging transport mehhanism.
  25
  26                                 To "hide" internal functions the choice was made to implement them
  27                                 all as static functions. This means that we include nearly
  28                                 all of our modules here as 90% of the library is not visible to
  29                                 the outside world.
  30
  31         Author:         E. Scott Daniels
  32         Date:           28 November 2018
  33 */
  34
  35 #include <ctype.h>
  36 #include <stdio.h>
  37 #include <stdlib.h>
  38 #include <netdb.h>
  39 #include <errno.h>
  40 #include <string.h>
  41 #include <errno.h>
  42 #include <pthread.h>
  43 #include <unistd.h>
  44 #include <stdint.h>
  45 #include <time.h>
  46 #include <arpa/inet.h>
  47
  48 #include <nanomsg/nn.h>
  49 #include <nanomsg/tcp.h>
  50 #include <nanomsg/pair.h>
  51 #include <nanomsg/pipeline.h>
  52 #include <nanomsg/pubsub.h>
  53
  54 #include "rmr.h"                                // things the users see
  55 #include "rmr_agnostic.h"               // headers agnostic to the underlying transport mechanism
  56 #include "rmr_private.h"                // things that we need too
  57 #include "rmr_symtab.h"
  58
  59 #include "ring_static.c"                // message ring support
  60 #include "rt_generic_static.c"  // generic route table (not nng/nano specific)
  61 #include "rtable_static.c"              // route table things   (nano specific)
  62 #include "rtc_static.c"                 // common rt collector
  63 #include "tools_static.c"
  64 #include "sr_static.c"                  // send/receive static functions
  65 #include "wormholes.c"                  // external wormhole api, and it's static functions (must be LAST)
  66
  67 // ------------------------------------------------------------------------------------------------------
  68
  69 /*
  70         Clean up a context.
  71 */
  72 static void free_ctx( uta_ctx_t* ctx ) {
  73         if( ctx ) {
  74                 if( ctx->rtg_addr ) {
  75                         free( ctx->rtg_addr );
  76                 }
  77         }
  78 }
  79
  80 // --------------- public functions --------------------------------------------------------------------------
  81
  82 /*
  83         Set the receive timeout to time (ms). A value of 0 is the same as a non-blocking
  84         receive and -1 is block for ever.
  85         Returns the nn value (0 on success <0 on error).
  86 */
  87 extern int rmr_set_rtimeout( void* vctx, int time ) {
  88         uta_ctx_t* ctx;
  89
  90         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
  91                 errno = EINVAL;
  92                 return -1;
  93         }
  94
  95         if( ctx->last_rto == time ) {
  96                 return 0;
  97         }
  98
  99         ctx->last_rto = time;
 100
 101         return nn_setsockopt( ctx->nn_sock, NN_SOL_SOCKET, NN_RCVTIMEO, &time, sizeof( time ) );
 102 }
 103
 104 /*
 105         Deprecated -- use rmr_set_rtimeout()
 106 */
 107 extern int rmr_rcv_to( void* vctx, int time ) {
 108         return rmr_rcv_to( vctx, time );
 109 }
 110
 111 /*
 112         Set the send timeout to time. If time >1000 we assume the time is milliseconds,
 113         else we assume seconds. Setting -1 is always block.
 114         Returns the nn value (0 on success <0 on error).
 115 */
 116 extern int rmr_set_stimeout( void* vctx, int time ) {
 117         uta_ctx_t* ctx;
 118
 119         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 120                 errno = EINVAL;
 121                 return -1;
 122         }
 123
 124         if( time > 0 ) {
 125                 if( time < 1000 ) {
 126                         time = time * 1000;                     // assume seconds, nn wants ms
 127                 }
 128         }
 129
 130         return nn_setsockopt( ctx->nn_sock, NN_SOL_SOCKET, NN_SNDTIMEO, &time, sizeof( time ) );
 131 }
 132
 133 /*
 134         Deprecated -- use rmr_set_stimeout()
 135 */
 136 extern int rmr_send_to( void* vctx, int time ) {
 137         return rmr_send_to( vctx, time );
 138 }
 139
 140 /*
 141         Returns the size of the payload (bytes) that the msg buffer references.
 142         Len in a message is the number of bytes which were received, or should
 143         be transmitted, however, it is possible that the mbuf was allocated
 144         with a larger payload space than the payload length indicates; this
 145         function returns the absolute maximum space that the user has available
 146         in the payload. On error (bad msg buffer) -1 is returned and errno should
 147         indicate the rason.
 148 */
 149 extern int rmr_payload_size( rmr_mbuf_t* msg ) {
 150         if( msg == NULL || msg->header == NULL ) {
 151                 errno = EINVAL;
 152                 return -1;
 153         }
 154
 155         errno = 0;
 156         return msg->alloc_len - RMR_HDR_LEN( msg->header );                     // transport buffer less header and other data bits
 157 }
 158
 159 /*
 160         Allocates a send message as a zerocopy message allowing the underlying message protocol
 161         to send the buffer without copy.
 162 */
 163 extern rmr_mbuf_t* rmr_alloc_msg( void* vctx, int size ) {
 164         uta_ctx_t*      ctx;
 165         rmr_mbuf_t*     m;
 166
 167         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 168                 return NULL;
 169         }
 170
 171         m = alloc_zcmsg( ctx, NULL, size, 0, DEF_TR_LEN );
 172         return  m;
 173 }
 174
 175 /*
 176         Allocates a send message as a zerocopy message allowing the underlying message protocol
 177         to send the buffer without copy. In addition, a trace data field of tr_size will be
 178         added and the supplied data coppied to the buffer before returning the message to
 179         the caller.
 180 */
 181 extern rmr_mbuf_t* rmr_tralloc_msg( void* vctx, int size, int tr_size, unsigned const char* data ) {
 182         uta_ctx_t*      ctx;
 183         rmr_mbuf_t*     m;
 184         int state;
 185
 186         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 187                 return NULL;
 188         }
 189
 190         m = alloc_zcmsg( ctx, NULL, size, 0, tr_size );                         // alloc with specific tr size
 191         if( m != NULL ) {
 192                 state = rmr_set_trace( m, data, tr_size );                              // roll their data in
 193                 if( state != tr_size ) {
 194                         m->state = RMR_ERR_INITFAILED;
 195                 }
 196         }
 197
 198         return  m;
 199 }
 200
 201 /*
 202         Need an external path to the realloc static function as it's called by an
 203         outward facing mbuf api function.
 204 */
 205 extern rmr_mbuf_t* rmr_realloc_msg( rmr_mbuf_t* msg, int new_tr_size ) {
 206         return realloc_msg( msg, new_tr_size );
 207 }
 208
 209 /*
 210         Return the message to the available pool, or free it outright.
 211 */
 212 extern void rmr_free_msg( rmr_mbuf_t* mbuf ) {
 213         if( mbuf == NULL ) {
 214                 return;
 215         }
 216
 217         if( mbuf->header ) {
 218                 if( mbuf->flags & MFL_ZEROCOPY ) {
 219                         nn_freemsg( mbuf->header );                             // must let nano free it
 220                 } else {
 221                         free( mbuf->header );
 222                 }
 223         }
 224
 225         free( mbuf );
 226 }
 227
 228 /*
 229         Accept a message and send it to an endpoint based on message type.
 230         Allocates a new message buffer for the next send. If a message type has
 231         more than one group of endpoints defined, then the message will be sent
 232         in round robin fashion to one endpoint in each group.
 233
 234         CAUTION: this is a non-blocking send.  If the message cannot be sent, then
 235                 it will return with an error and errno set to eagain. If the send is
 236                 a limited fanout, then the returned status is the status of the last
 237                 send attempt.
 238 */
 239 extern rmr_mbuf_t* rmr_send_msg( void* vctx, rmr_mbuf_t* msg ) {
 240         int nn_sock;                            // endpoint socket for send
 241         uta_ctx_t*      ctx;
 242         int     group;                                  // selected group to get socket for
 243         int send_again;                         // true if the message must be sent again
 244         rmr_mbuf_t*     clone_m;                // cloned message for an nth send
 245
 246         if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
 247                 errno = EINVAL;                                                                                         // if msg is null, this is their clue
 248                 if( msg != NULL ) {
 249                         msg->state = RMR_ERR_BADARG;
 250                         errno = EINVAL;                                                                                 // must ensure it's not eagain
 251                 }
 252                 return msg;
 253         }
 254
 255         errno = 0;                                                                                                      // clear; nano might set, but ensure it's not left over if it doesn't
 256         if( msg->header == NULL ) {
 257                 fprintf( stderr, "[ERR] rmr_send_msg: message had no header\n" );
 258                 msg->state = RMR_ERR_NOHDR;
 259                 errno = EBADMSG;                                                                                // must ensure it's not eagain
 260                 return msg;
 261         }
 262
 263         send_again = 1;                                                                                 // force loop entry
 264         group = 0;                                                                                              // always start with group 0
 265
 266         while( send_again ) {
 267                 nn_sock = uta_epsock_rr( ctx->rtable, msg->mtype, group, &send_again );         // round robin select endpoint; again set if mult groups
 268                 if( DEBUG ) fprintf( stderr, "[DBUG] send msg: type=%d again=%d group=%d socket=%d len=%d\n",
 269                                 msg->mtype, send_again, group, nn_sock, msg->len );
 270                 group++;
 271
 272                 if( nn_sock < 0 ) {
 273                         msg->state = RMR_ERR_NOENDPT;
 274                         errno = ENXIO;                                                                                  // must ensure it's not eagain
 275                         return msg;                                                                                             // caller can resend (maybe) or free
 276                 }
 277
 278                 if( send_again ) {
 279                         clone_m = clone_msg( msg );                                                             // must make a copy as once we send this message is not available
 280                         if( DEBUG ) fprintf( stderr, "[DBUG] msg cloned: type=%d len=%d\n", msg->mtype, msg->len );
 281                         msg->flags |= MFL_NOALLOC;                                                              // send should not allocate a new buffer
 282                         msg = send_msg( ctx, msg, nn_sock );                                    // do the hard work, msg should be nil on success
 283                         /*
 284                         if( msg ) {
 285                                 // error do we need to count successes/errors, how to report some success, esp if last fails?
 286                         }
 287                         */
 288
 289                         msg = clone_m;                                                                                  // clone will be the next to send
 290                 } else {
 291                         msg = send_msg( ctx, msg, nn_sock );                                    // send the last, and allocate a new buffer; drops the clone if it was
 292                 }
 293         }
 294
 295         return msg;                                                                     // last message caries the status of last/only send attempt
 296 }
 297
 298 /*
 299         Return to sender allows a message to be sent back to the endpoint where it originated.
 300         The source information in the message is used to select the socket on which to write
 301         the message rather than using the message type and round-robin selection. This
 302         should return a message buffer with the state of the send operation set. On success
 303         (state is RMR_OK, the caller may use the buffer for another receive operation), and on
 304         error it can be passed back to this function to retry the send if desired. On error,
 305         errno will liklely have the failure reason set by the nanomsg send processing.
 306         The following are possible values for the state in the message buffer:
 307
 308         Message states returned:
 309                 RMR_ERR_BADARG - argument (context or msg) was nil or invalid
 310                 RMR_ERR_NOHDR  - message did not have a header
 311                 RMR_ERR_NOENDPT- an endpoint to send the message to could not be determined
 312                 RMR_ERR_SENDFAILED - send failed; errno has nano error code
 313                 RMR_ERR_RETRY   - operation failed, but caller should retry
 314
 315         A nil message as the return value is rare, and generally indicates some kind of horrible
 316         failure. The value of errno might give a clue as to what is wrong.
 317
 318         CAUTION:
 319                 Like send_msg(), this is non-blocking and will return the msg if there is an errror.
 320                 The caller must check for this and handle.
 321 */
 322 extern rmr_mbuf_t*  rmr_rts_msg( void* vctx, rmr_mbuf_t* msg ) {
 323         int nn_sock;                            // endpoint socket for send
 324         uta_ctx_t*      ctx;
 325         int state;
 326         uta_mhdr_t*     hdr;
 327         char*   hold_src;                       // we need the original source if send fails
 328
 329         if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
 330                 errno = EINVAL;                                                                                         // if msg is null, this is their clue
 331                 if( msg != NULL ) {
 332                         msg->state = RMR_ERR_BADARG;
 333                 }
 334                 return msg;
 335         }
 336
 337         errno = 0;                                                                                                              // at this point any bad state is in msg returned
 338         if( msg->header == NULL ) {
 339                 fprintf( stderr, "rmr_send_msg: ERROR: message had no header\n" );
 340                 msg->state = RMR_ERR_NOHDR;
 341                 return msg;
 342         }
 343
 344         nn_sock = uta_epsock_byname( ctx->rtable, (char *) ((uta_mhdr_t *)msg->header)->src );                  // socket of specific endpoint
 345         if( nn_sock < 0 ) {
 346                 msg->state = RMR_ERR_NOENDPT;
 347                 return msg;                                                     // preallocated msg can be reused since not given back to nn
 348         }
 349
 350         hold_src = strdup( (char *) ((uta_mhdr_t *)msg->header)->src );                                                 // the dest where we're returning the message to
 351         strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SID );                // must overlay the source to be ours
 352         msg = send_msg( ctx, msg, nn_sock );
 353         if( msg ) {
 354                 strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, hold_src, RMR_MAX_SID );            // always return original source so rts can be called again
 355                 msg->flags |= MFL_ADDSRC;                                                                                                       // if msg given to send() it must add source
 356         }
 357
 358         free( hold_src );
 359         return msg;
 360 }
 361
 362 /*
 363         Call sends the message based on message routing using the message type, and waits for a
 364         response message to arrive with the same transaction id that was in the outgoing message.
 365         If, while wiating for the expected response,  messages are received which do not have the
 366         desired transaction ID, they are queued. Calls to uta_rcv_msg() will dequeue them in the
 367         order that they were received.
 368
 369         Normally, a message struct pointer is returned and msg->state must be checked for RMR_OK
 370         to ensure that no error was encountered. If the state is UTA_BADARG, then the message
 371         may be resent (likely the context pointer was nil).  If the message is sent, but no
 372         response is received, a nil message is returned with errno set to indicate the likley
 373         issue:
 374                 ETIMEDOUT -- too many messages were queued before reciving the expected response
 375                 ENOBUFS -- the queued message ring is full, messages were dropped
 376                 EINVAL  -- A parameter was not valid
 377                 EAGAIN  -- the underlying message system wsa interrupted or the device was busy;
 378                                         user should call this function with the message again.
 379
 380
 381         QUESTION:  should user specify the number of messages to allow to queue?
 382 */
 383 extern rmr_mbuf_t* rmr_call( void* vctx, rmr_mbuf_t* msg ) {
 384         uta_ctx_t*              ctx;
 385         unsigned char   expected_id[RMR_MAX_XID+1];             // the transaction id in the message; we wait for response with same ID
 386
 387         if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
 388                 if( msg != NULL ) {
 389                         msg->state = RMR_ERR_BADARG;
 390                 }
 391                 return msg;
 392         }
 393
 394         memcpy( expected_id, msg->xaction, RMR_MAX_XID );
 395         expected_id[RMR_MAX_XID] = 0;                                   // ensure it's a string
 396         if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rmr_call is making call, waiting for (%s)\n", expected_id );
 397         errno = 0;
 398         msg->flags |= MFL_NOALLOC;                                              // we don't need a new buffer from send
 399
 400         msg = rmr_send_msg( ctx, msg );
 401         if( msg ) {                                                                             // msg should be nil, if not there was a problem; return buffer to user
 402                 if( msg->state != RMR_ERR_RETRY ) {
 403                         msg->state = RMR_ERR_CALLFAILED;                // don't stomp if send_msg set retry
 404                 }
 405                 return msg;
 406         }
 407
 408         return rmr_rcv_specific( ctx, NULL, (char *) expected_id, 20 );                 // wait for msg allowing 20 to queue ahead
 409 }
 410
 411 /*
 412         The outward facing receive function. When invoked it will pop the oldest message
 413         from the receive ring, if any are queued, and return it. If the ring is empty
 414         then the receive function is invoked to wait for the next message to arrive (blocking).
 415
 416         If old_msg is provided, it will be populated (avoiding lots of free/alloc cycles). If
 417         nil, a new one will be allocated. However, the caller should NOT expect to get the same
 418         struct back (if a queued message is returned the message struct will be different).
 419 */
 420 extern rmr_mbuf_t* rmr_rcv_msg( void* vctx, rmr_mbuf_t* old_msg ) {
 421         uta_ctx_t*      ctx;
 422         rmr_mbuf_t*     qm;                             // message that was queued on the ring
 423
 424         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 425                 if( old_msg != NULL ) {
 426                         old_msg->state = RMR_ERR_BADARG;
 427                 }
 428                 errno = EINVAL;
 429                 return old_msg;
 430         }
 431         errno = 0;
 432
 433         qm = (rmr_mbuf_t *) uta_ring_extract( ctx->mring );                     // pop if queued
 434         if( qm != NULL ) {
 435                 if( old_msg ) {
 436                         rmr_free_msg( old_msg );                                                        // future:  push onto a free list???
 437                 }
 438
 439                 return qm;
 440         }
 441
 442         return rcv_msg( ctx, old_msg );                                                         // nothing queued, wait for one
 443 }
 444
 445 /*
 446         Receive with a timeout.  This is a convenience function when sitting on top of
 447         nanomsg as it just sets the rcv timeout and calls rmr_rcv_msg().
 448 */
 449 extern rmr_mbuf_t* rmr_torcv_msg( void* vctx, rmr_mbuf_t* old_msg, int ms_to ) {
 450         uta_ctx_t*      ctx;
 451
 452         if( (ctx = (uta_ctx_t *) vctx) != NULL ) {
 453                 if( ctx->last_rto != ms_to ) {                                                  // avoid call overhead
 454                         rmr_set_rtimeout( vctx, ms_to );
 455                 }
 456         }
 457
 458         return rmr_rcv_msg( vctx, old_msg );
 459 }
 460
 461
 462 /*
 463         This blocks until the message with the 'expect' ID is received. Messages which are received
 464         before the expected message are queued onto the message ring.  The function will return
 465         a nil message and set errno to ETIMEDOUT if allow2queue messages are received before the
 466         expected message is received. If the queued message ring fills a nil pointer is returned
 467         and errno is set to ENOBUFS.
 468
 469         Generally this will be invoked only by the call() function as it waits for a response, but
 470         it is exposed to the user application as three is no reason not to.
 471 */
 472 extern rmr_mbuf_t* rmr_rcv_specific( void* vctx, rmr_mbuf_t* msg, char* expect, int allow2queue ) {
 473         uta_ctx_t*      ctx;
 474         int     queued = 0;                             // number we pushed into the ring
 475         int     exp_len = 0;                    // length of expected ID
 476
 477         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 478                 if( msg != NULL ) {
 479                         msg->state = RMR_ERR_BADARG;
 480                 }
 481                 errno = EINVAL;
 482                 return msg;
 483         }
 484
 485         errno = 0;
 486
 487         if( expect == NULL || ! *expect ) {                             // nothing expected if nil or empty string, just receive
 488                 return rmr_rcv_msg( ctx, msg );
 489         }
 490
 491         exp_len = strlen( expect );
 492         if( exp_len > RMR_MAX_XID ) {
 493                 exp_len = RMR_MAX_XID;
 494         }
 495         if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific waiting for id=%s\n",  expect );
 496
 497         while( queued < allow2queue ) {
 498                 msg = rcv_msg( ctx, msg );                                      // hard wait for next
 499                 if( msg->state == RMR_OK ) {
 500                         if( memcmp( msg->xaction, expect, exp_len ) == 0 ) {                    // got it -- return it
 501                                 if( DEBUG ) fprintf( stderr, "[DBUG] rcv-specific matched (%s); %d messages were queued\n", msg->xaction, queued );
 502                                 return msg;
 503                         }
 504
 505                         if( ! uta_ring_insert( ctx->mring, msg ) ) {                                    // just queue, error if ring is full
 506                                 if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_specific ring is full\n" );
 507                                 errno = ENOBUFS;
 508                                 return NULL;
 509                         }
 510
 511                         if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific queued message type=%d\n", msg->mtype );
 512                         queued++;
 513                         msg = NULL;
 514                 }
 515         }
 516
 517         if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific timeout waiting for %s\n", expect );
 518         errno = ETIMEDOUT;
 519         return NULL;
 520 }
 521
 522
 523 /*
 524         Initialise the message routing environment. Flags are one of the UTAFL_
 525         constants. Proto_port is a protocol:port string (e.g. tcp:1234). If default protocol
 526         (tcp) to be used, then :port is all that is needed.
 527
 528         At the moment it seems that TCP really is the only viable protocol, but
 529         we'll allow flexibility.
 530
 531         The return value is a void pointer which must be passed to most uta functions. On
 532         error, a nil pointer is returned and errno should be set.
 533 */
 534 static void* init( char* uproto_port, int max_msg_size, int flags ) {
 535         uta_ctx_t*      ctx = NULL;
 536         char    bind_info[NN_SOCKADDR_MAX];     // bind info
 537         char*   proto = "tcp";                          // pointer into the proto/port string user supplied
 538         char*   port;
 539         char*   proto_port;
 540         char    wbuf[1024];                                     // work buffer
 541         char*   tok;                                            // pointer at token in a buffer
 542         int             state;
 543         char*   interface = NULL;                       // interface to bind to pulled from RMR_BIND_IF if set
 544
 545         fprintf( stderr, "[INFO] ric message routing library on nanomsg (%s %s.%s.%s built: %s)\n",
 546                         QUOTE_DEF(GIT_ID), QUOTE_DEF(MAJOR_VER), QUOTE_DEF(MINOR_VER), QUOTE_DEF(PATCH_VER), __DATE__ );
 547
 548         errno = 0;
 549         if( uproto_port == NULL ) {
 550                 proto_port = strdup( "tcp:4567" );
 551         } else {
 552                 proto_port = strdup( uproto_port );             // so we can modify it
 553         }
 554
 555         if( (ctx = (uta_ctx_t *) malloc( sizeof( uta_ctx_t ) )) == NULL ) {
 556                 errno = ENOMEM;
 557                 return NULL;
 558         }
 559         memset( ctx, 0, sizeof( uta_ctx_t ) );
 560
 561
 562         ctx->mring = uta_mk_ring( 128 );                                // message ring to hold asynch msgs received while waiting for call response
 563         ctx->last_rto = -2;                                                             // last receive timeout that was set; invalid value to force first to set
 564
 565         ctx->max_plen = RMR_MAX_RCV_BYTES + sizeof( uta_mhdr_t );               // default max buffer size
 566         if( max_msg_size > 0 ) {
 567                 if( max_msg_size <= ctx->max_plen ) {                                           // user defined len can be smaller
 568                         ctx->max_plen = max_msg_size;
 569                 } else {
 570                         fprintf( stderr, "[WARN] rmr_init: attempt to set max payload len > than allowed maximum; capped at %d bytes\n", ctx->max_plen );
 571                 }
 572         }
 573
 574         ctx->max_mlen = ctx->max_plen + sizeof( uta_mhdr_t );
 575
 576         uta_lookup_rtg( ctx );                                                  // attempt to fill in rtg info; rtc will handle missing values/errors
 577
 578         ctx->nn_sock = nn_socket( AF_SP, NN_PULL );             // our 'listen' socket should allow multiple senders to connect
 579         if( ctx->nn_sock < 0 ) {
 580                 fprintf( stderr, "[CRIT] rmr_init: unable to initialise nanomsg listen socket: %d\n", errno );
 581                 free_ctx( ctx );
 582                 return NULL;
 583         }
 584
 585         if( (port = strchr( proto_port, ':' )) != NULL ) {
 586                 if( port == proto_port ) {              // ":1234" supplied; leave proto to default and point port correctly
 587                         port++;
 588                 } else {
 589                         *(port++) = 0;                  // term proto string and point at port string
 590                         proto = proto_port;             // user supplied proto so point at it rather than default
 591                 }
 592         } else {
 593                 port = proto_port;                      // assume something like "1234" was passed
 594         }
 595
 596         if( (gethostname( wbuf, sizeof( wbuf ) )) < 0 ) {
 597                 fprintf( stderr, "[CRIT] rmr_init: cannot determine localhost name: %s\n", strerror( errno ) );
 598                 return NULL;
 599         }
 600         if( (tok = strchr( wbuf, '.' )) != NULL ) {
 601                 *tok = 0;                                                                       // we don't keep domain portion
 602         }
 603         ctx->my_name = (char *) malloc( sizeof( char ) * RMR_MAX_SID );
 604         if( snprintf( ctx->my_name, RMR_MAX_SID, "%s:%s", wbuf, port ) >= RMR_MAX_SID ) {                       // our registered name is host:port
 605                 fprintf( stderr, "[CRIT] rmr_init: hostname + port must be less than %d characters; %s:%s is not\n", RMR_MAX_SID, wbuf, port );
 606                 return NULL;
 607         }
 608
 609         if( (interface = getenv( ENV_BIND_IF )) == NULL ) {
 610                 interface = "0.0.0.0";
 611         }
 612         snprintf( bind_info, sizeof( bind_info ), "%s://%s:%s", proto, interface, port );
 613         if( nn_bind( ctx->nn_sock, bind_info ) < 0) {                   // bind and automatically accept client sessions
 614                 fprintf( stderr, "[CRIT] rmr_init: unable to bind nanomsg listen socket for %s: %s\n", bind_info, strerror( errno ) );
 615                 nn_close( ctx->nn_sock );
 616                 free_ctx( ctx );
 617                 return NULL;
 618         }
 619
 620         if( ! (flags & FL_NOTHREAD) ) {                 // skip if internal context that does not need rout table thread
 621                 if( pthread_create( &ctx->rtc_th,  NULL, rtc, (void *) ctx ) ) {                // kick the rt collector thread
 622                         fprintf( stderr, "[WARN] rmr_init: unable to start route table collector thread: %s", strerror( errno ) );
 623                 }
 624         }
 625
 626         free( proto_port );
 627         return (void *) ctx;
 628 }
 629
 630
 631 /*
 632         Publicly facing initialisation function. Wrapper for the init() funcion above
 633         as it needs to ensure internal flags are masked off before calling the
 634         real workhorse.
 635 */
 636 extern void* rmr_init( char* uproto_port, int max_msg_size, int flags ) {
 637         return init( uproto_port, max_msg_size, flags & UFL_MASK  );
 638 }
 639
 640 /*
 641         Return true if routing table is initialised etc. and app can send/receive.
 642 */
 643 extern int rmr_ready( void* vctx ) {
 644         uta_ctx_t *ctx;
 645
 646         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 647                 return FALSE;
 648         }
 649
 650         if( ctx->rtable != NULL ) {
 651                 return TRUE;
 652         }
 653
 654         return FALSE;
 655 }
 656
 657 /*
 658         Provides a non-fatal (compile) interface for the nng only function.
 659         Not supported on top of nano, so this always returns -1.
 660 */
 661 extern int rmr_get_rcvfd( void* vctx ) {
 662         errno = ENOTSUP;
 663         return -1;
 664 }
 665
 666 /*
 667         Compatability (mostly) with NNG.
 668 */
 669 extern void rmr_close( void* vctx ) {
 670         uta_ctx_t *ctx;
 671
 672         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 673                 return;
 674         }
 675
 676         nn_close( ctx->nn_sock );
 677 }