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