src/rmr/common/src/rtc_static.c

   1 // : vi ts=4 sw=4 noet :
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2019-2020 Nokia
   5         Copyright (c) 2018-2020 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:       rt_collector.c
  23         Abstract:       The route table collector is started as a separate pthread and
  24                                 is responsible for listening for route table updates from a
  25                                 route manager or route table generator process.
  26
  27         Author:         E. Scott Daniels
  28         Date:           29 November 2018 (extracted to common 13 March 2019)
  29 */
  30
  31 #ifndef _rt_collector_c
  32 #define _rt_collector_c
  33
  34 #include <stdio.h>
  35 #include <stdlib.h>
  36 #include <netdb.h>
  37 #include <errno.h>
  38 #include <string.h>
  39 #include <errno.h>
  40 #include <fcntl.h>
  41 #include <sys/types.h>
  42 #include <sys/stat.h>
  43 #include <unistd.h>
  44
  45 #include <RIC_message_types.h>          // needed for RMR/Rt Mgr msg types
  46
  47 // ---- local constants ------------------
  48                                                                                 // flags
  49 #define RTCFL_HAVE_UPDATE       0x01            // an update from RM was received
  50
  51 #define MAX_RTC_BUF     5 * 1024                        // max buffer size we'll expect is 4k, add some fudge room
  52
  53 // ------------------------------------------------------------------------------------------------
  54
  55 /*
  56         Loop forever (assuming we're running in a pthread reading the static table
  57         every minute or so.
  58 */
  59 static void* rtc_file( void* vctx ) {
  60         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
  61         char*   eptr;
  62         int             vfd = -1;                                       // verbose file des if we have one
  63         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
  64         char    wbuf[256];
  65
  66
  67         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
  68                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
  69                 return NULL;
  70         }
  71
  72         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
  73                 vfd = open( eptr, O_RDONLY );
  74         }
  75
  76         ctx->flags |= CFL_NO_RTACK;                             // no attempt to ack when reading from a file
  77         while( 1 ) {
  78                 if( vfd >= 0 ) {
  79                         wbuf[0] = 0;
  80                         lseek( vfd, 0, 0 );
  81                         read( vfd, wbuf, 10 );
  82                         vlevel = atoi( wbuf );
  83                 }
  84
  85                 read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
  86
  87                 sleep( 60 );
  88         }
  89 }
  90
  91 static int refresh_vlevel( int vfd ) {
  92         int vlevel = 0;
  93         char    rbuf[128];
  94
  95         if( vfd >= 0 ) {                                        // if file is open, read current value
  96                 rbuf[0] = 0;
  97                 lseek( vfd, 0, 0 );
  98                 read( vfd, rbuf, 10 );
  99                 vlevel = atoi( rbuf );
 100         }
 101
 102         return vlevel;
 103 }
 104
 105 /*
 106         Route Table Collector
 107         A side thread which either attempts to connect and request a table
 108         from the Route Manager, or opens a port and listens for Route Manager
 109         to push table updates.
 110
 111         It may do other things along the way (latency measurements, alarms,
 112         respond to RMR pings, etc.).
 113
 114         The behaviour with respect to listening for Route Manager updates vs
 115         the initiation of the connection and sending a request depends on the
 116         value of the ENV_RTG_ADDR (RMR_RTG_SVC)  environment variable. If
 117         host:port, or IP:port, is given, then we assume that we make the connection
 118         and send a request for the table (request mode).  If the variable is just
 119         a port, then we assume Route Manager will connect and push updates (original
 120         method).
 121
 122         If the variable is not defined, the default behaviour, in order to be
 123         backwards compatable, depends on the presence of the ENV_CTL_PORT
 124         (RMR_CTL_PORT) variable (new with the support for requesting a table).
 125
 126
 127         ENV_CTL_PORT    ENV_RTG_ADDR    Behaviour
 128         unset                   unset                   Open default CTL port (DEF_CTL_PORT) and
 129                                                                         wait for Rt Mgr to push tables
 130
 131         set                             unset                   Use the default Rt Mgr wellknown addr
 132                                                                         and port (DEF_RTG_WK_ADDR) to connect
 133                                                                         and request a table. The control port
 134                                                                         used is the value set by ENV_CTL_PORT.
 135
 136         unset                   set                             As described above. The default control
 137                                                                         port (DEF_CTL_PORT) is used.
 138
 139         When we are running in request mode, then we will send the RMR message
 140         RMRRM_REFRESH to this address (wormhole) as a request for the route manager
 141         to send a new table. We will attempt to connect and send requests until
 142         we have a table. Calls to rmr_ready() will report FALSE until a table is
 143         loaded _unless_ a seed table was given.
 144
 145         Route table information is expected to arrive on RMR messages with type
 146         RMRRM_TABLE_DATA.  There is NOT a specific message type for each possible
 147         table record, so the payload is as it appears in the seed file or as
 148         delivered in old versions.  It may take several RMRRM_TABLE_DATA messages
 149         to completely supply a new table or table update. See the header for parse_rt_rec
 150         in common for a description of possible message contents.
 151
 152         Buffers received from the route table generator can contain multiple newline terminated
 153         records, but each buffer must be less than 4K in length, and the last record in a
 154         buffer may NOT be split across buffers.
 155
 156         Other chores:
 157         In addition to the primary task of getting, vetting, and installing a new route table, or
 158         updates to the existing table, this thread will periodically cause the send counts for each
 159         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
 160         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
 161 */
 162 static void* rtc( void* vctx ) {
 163         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
 164         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
 165         rmr_mbuf_t*     msg = NULL;                             // message from rtg
 166         char*   payload;                                        // payload in the message
 167         size_t  mlen;
 168         char*   my_port;                                        // the port number that we will listen on (4561 has been the default for this)
 169         char*   rtg_addr;                                       // host:port address of route table generator (route manager)
 170         char*   daddr;                                          // duplicated rtg address string to parse/trash
 171         size_t  buf_size;                                       // nng needs var pointer not just size?
 172         char*   nextr;                                          // pointer at next record in the message
 173         char*   curr;                                           // current record
 174         int     i;
 175         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
 176         int             cstate = -1;                            // connection state to rtg
 177         int             state;                                          // processing state of some nng function
 178         char*   tokens[128];
 179         char    wbuf[128];
 180         char*   pbuf = NULL;
 181         int             pbuf_size = 0;                          // number allocated in pbuf
 182         int             ntoks;
 183         int             vfd = -1;                                       // verbose file des if we have one
 184         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 185         char*   eptr;
 186         int             epfd = -1;                                      // fd for epoll so we can multi-task
 187         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 188         struct  epoll_event epe;                        // event definition for event to listen to
 189         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 190         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 191         int             flags = 0;
 192
 193
 194         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 195                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 196                 return NULL;
 197         }
 198
 199         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 200                 vfd = open( eptr, O_RDONLY );
 201                 vlevel = refresh_vlevel( vfd );
 202         }
 203
 204         ctx->flags |= CFL_NO_RTACK;                     // don't ack when reading from a file
 205         read_static_rt( ctx, vlevel );                  // seed the route table if one provided
 206         ctx->flags &= ~CFL_NO_RTACK;
 207
 208
 209         my_port = getenv( ENV_CTL_PORT );                               // default port to listen on (likely 4561)
 210         if( my_port == NULL || ! *my_port ) {                   // if undefined, then go with default
 211                 my_port = DEF_CTL_PORT;
 212                 daddr = DEF_CTL_PORT;                                           // backwards compat; if ctl port not hard defined, default is to listen
 213         } else {
 214                 daddr = DEF_RTG_WK_ADDR;                                        // if ctl port is defined, then default changes to connecting to well known RM addr
 215         }
 216
 217         if( (rtg_addr = getenv( ENV_RTG_ADDR )) == NULL || ! *rtg_addr ) {              // undefined, use default set above
 218                 rtg_addr = daddr;
 219         }
 220
 221         daddr = strdup( rtg_addr );                                                                     // dup to destroy during parse
 222
 223         ntoks = uta_tokenise( daddr, tokens, 120, ':' );                        // should be host:ip of rt mgr (could be port only which we assume is old listen port)
 224         switch( ntoks ) {
 225                 case 0:                                                                 // should not happen, but prevent accidents and allow default to ignore additional tokens
 226                         break;
 227
 228                 case 1:
 229                         my_port = tokens[0];                    // just port -- assume backlevel environment where we just listen
 230                         flags |= RTCFL_HAVE_UPDATE;             // prevent sending update reqests
 231                         break;
 232
 233                 default:
 234                         if( strcmp( tokens[0], "tcp" ) == 0 ) {                 // old school nng tcp:xxxx so we listen on xxx
 235                                 flags |= RTCFL_HAVE_UPDATE;                                     // and signal not to try to request an update
 236                                 my_port = tokens[1];
 237                         } else {
 238                                 // rtg_addr points at rt mgr address and my port set from env or default stands as is
 239                         }
 240                         break;
 241         }
 242
 243         if( (pvt_cx = init( my_port, MAX_RTC_BUF, FL_NOTHREAD )) == NULL ) {                            // open a private context (no RT listener!)
 244                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 245
 246                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 247                         sleep( count_delay );
 248                         rt_epcounts( ctx->rtable, ctx->my_name );
 249                 }
 250
 251                 return NULL;
 252         }
 253
 254         ctx->rtg_whid = -1;
 255
 256         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", my_port );
 257
 258         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 259         blabber = 0;
 260         while( 1 ) {                                                                                                            // until the cows return, pigs fly, or somesuch event likely not to happen
 261                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 262                         if( (flags & RTCFL_HAVE_UPDATE) == 0 ) {                                                // no route table updated from rt mgr; request one
 263                                 if( ctx->rtg_whid < 0 ) {
 264                                         ctx->rtg_whid = rmr_wh_open( pvt_cx, rtg_addr );
 265                                 }
 266                                 send_update_req( pvt_cx, ctx );
 267                         }
 268
 269                         msg = rmr_torcv_msg( pvt_cx, msg, 1000 );
 270
 271                         if( time( NULL ) > blabber  ) {
 272                                 vlevel = refresh_vlevel( vfd );
 273                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 274                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 275                                         if( blabber > bump_freq ) {
 276                                                 count_delay = 300;
 277                                         }
 278                                         rt_epcounts( ctx->rtable, ctx->my_name );
 279                                 }
 280                         }
 281                 }
 282
 283                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 284
 285                 if( msg != NULL && msg->len > 0 ) {
 286                         payload = msg->payload;
 287                         mlen = msg->len;                                        // usable bytes in the payload
 288
 289                         if( DEBUG > 1 || (vlevel > 0) ) rmr_vlog( RMR_VL_DEBUG, "rmr_rtc: received rt message type=%d len=%d\n", msg->mtype, (int) mlen );
 290                         switch( msg->mtype ) {
 291                                 case RMRRM_TABLE_DATA:
 292                                         if( (flags & RTCFL_HAVE_UPDATE) == 0 ) {
 293                                                 flags |= RTCFL_HAVE_UPDATE;
 294                                                 rmr_vlog( RMR_VL_INFO, "message flow from route manager starts\n" );
 295                                         }
 296
 297                                         if( pbuf_size <= mlen ) {
 298                                                 if( pbuf ) {
 299                                                         free( pbuf );
 300                                                 }
 301                                                 if( mlen < 512 ) {
 302                                                         pbuf_size = 512;
 303                                                 } else {
 304                                                         pbuf_size = mlen * 2;
 305                                                 }
 306                                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 307                                         }
 308                                         memcpy( pbuf, payload, mlen );
 309                                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 310                                         if( vlevel > 1 ) {
 311                                                 rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: rt message: (%s)\n", pbuf );
 312                                         }
 313
 314                                         curr = pbuf;
 315                                         while( curr ) {                                                         // loop over each record in the buffer
 316                                                 nextr = strchr( curr, '\n' );                   // allow multiple newline records, find end of current and mark
 317
 318                                                 if( nextr ) {
 319                                                         *(nextr++) = 0;
 320                                                 }
 321
 322                                                 if( vlevel > 1 ) {
 323                                                         rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: processing (%s)\n", curr );
 324                                                 }
 325                                                 parse_rt_rec( ctx, pvt_cx, curr, vlevel, msg );         // parse record and add to in progress table; ack using rts to msg
 326
 327                                                 curr = nextr;
 328                                         }
 329
 330                                         msg->len = 0;                           // force back into the listen loop
 331                                         break;
 332
 333                                 default:
 334                                         rmr_vlog( RMR_VL_WARN, "rmr_rtc: invalid message type=%d len=%d\n", msg->mtype, (int) msg->len );
 335                                         break;
 336                         }
 337                 }
 338
 339                 if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 340                         return NULL;
 341                 }
 342
 343         }
 344
 345         return NULL;    // unreachable, but some compilers don't see that and complain.
 346 }
 347
 348 #ifndef SI95_BUILD
 349 // this is nng specific inas much as we allow raw (non-RMR) messages
 350
 351 /*
 352         NOTE:   This is the original rtc code when we supported "raw" nano/nng messages
 353                         from the route manger.  It is deprecated in favour of managing all RM-RMR
 354                         communications via an RMR session.
 355
 356                         The rtc() function above is the new and preferred function regardless
 357                         of transport.
 358
 359         -----------------------------------------------------------------------------------
 360         Route Table Collector
 361         A side thread which opens a socket and subscribes to a routing table generator.
 362         It may do other things along the way (latency measurements?).
 363
 364         The pointer is a pointer to the context.
 365
 366         Listens for records from the route table generation publisher, expecting
 367         one of the following, newline terminated, ASCII records:
 368                 rte|msg-type||]name:port,name:port,...;name:port,...                    // route table entry with one or more groups of endpoints
 369                 new|start                                                               // start of new table
 370                 new|end                                                                 // end of new table; complete
 371
 372                 Name must be a host name which can be looked up via gethostbyname() (DNS).
 373
 374                 Multiple endpoints (name:port) may be given separated by a comma; an endpoint is selected using round robin
 375                         for each message of the type that is sent.
 376
 377                 Multiple endpoint groups can be given as a comma separated list of endpoints, separated by semicolons:
 378                                 group1n1:port,group1n2:port,group1n3:port;group2n1:port,group2n2:port
 379
 380                 If multiple groups are given, when send() is called for the cooresponding message type,
 381                 the message will be sent to one endpoint in each group.
 382
 383                 msg-type is the numeric message type (e.g. 102). If it is given as n,name then it is assumed
 384                 that the entry applies only to the instance running with the hostname 'name.'
 385
 386         Buffers received from the route table generator can contain multiple newline terminated
 387         records, but each buffer must be less than 4K in length, and the last record in a
 388         buffer may NOT be split across buffers.
 389
 390         Other chores:
 391         In addition to the primary task of getting, vetting, and installing a new route table, or
 392         updates to the existing table, this thread will periodically cause the send counts for each
 393         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
 394         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
 395 */
 396 static void* raw_rtc( void* vctx ) {
 397         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
 398         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
 399         rmr_mbuf_t*     msg = NULL;                             // message from rtg
 400         char*   payload;                                        // payload in the message
 401         size_t  mlen;
 402         char*   port;                                           // a port number we listen/connect to
 403         char*   fport;                                          // pointer to the real buffer to free
 404         size_t  buf_size;                                       // nng needs var pointer not just size?
 405         char*   nextr;                                          // pointer at next record in the message
 406         char*   curr;                                           // current record
 407         int     i;
 408         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
 409         int             cstate = -1;                            // connection state to rtg
 410         int             state;                                          // processing state of some nng function
 411         char*   tokens[128];
 412         char    wbuf[128];
 413         char*   pbuf = NULL;
 414         int             pbuf_size = 0;                          // number allocated in pbuf
 415         int             ntoks;
 416         int             raw_interface = 1;                      // rtg is using raw NNG/Nano not RMr to send updates
 417         int             vfd = -1;                                       // verbose file des if we have one
 418         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 419         char*   eptr;
 420         int             epfd = -1;                                      // fd for epoll so we can multi-task
 421         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 422         struct  epoll_event epe;                        // event definition for event to listen to
 423         int             rcv_fd = -1;                            // pollable file des from NNG to use for timeout
 424         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 425         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 426
 427
 428         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 429                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 430                 return NULL;
 431         }
 432
 433         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 434                 vfd = open( eptr, O_RDONLY );
 435                 vlevel = refresh_vlevel( vfd );
 436         }
 437
 438         read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
 439
 440         if( (port = getenv( ENV_RTG_PORT )) == NULL || ! *port ) {              // port we need to open to listen for RTG connections
 441                 port = strdup( DEF_RTG_PORT );
 442         } else {
 443                 port = strdup( port );
 444         }
 445
 446 /*
 447         this test is now done in init and this function is started _only_ if the value was 1
 448         if( (curr = getenv( ENV_RTG_RAW )) != NULL ) {
 449                 raw_interface = atoi( curr ) > 0;                               // if > 0 we assume that rtg messages are NOT coming from an RMr based process
 450         }
 451 */
 452
 453         fport = port;           // must hold to free
 454
 455         ntoks = uta_tokenise( port, tokens, 120, ':' );                 // assume tcp:port, but it could be port or old style host:port
 456         switch( ntoks ) {
 457                 case 1:
 458                                 port = tokens[0];                       // just the port
 459                                 break;
 460
 461                 case 2:
 462                                 port = tokens[1];                       // tcp:port or :port
 463                                 break;
 464
 465                 default:
 466                                 port = DEF_RTG_PORT;            // this shouldn't happen, but parnioia is good
 467                                 break;
 468         }
 469
 470         if( (pvt_cx = init( port, MAX_RTG_MSG_SZ, FL_NOTHREAD )) == NULL ) {                            // open a private context
 471                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 472
 473                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 474                         sleep( count_delay );
 475                         rt_epcounts( ctx->rtable, ctx->my_name );
 476                 }
 477
 478                 free( fport );                                  // parinoid free and return
 479                 return NULL;
 480         }
 481
 482         if( (rcv_fd = rmr_get_rcvfd( pvt_cx )) >= 0 ) {            // get the epoll fd for the rtg socket
 483                 if( rcv_fd < 0 ) {
 484                         rmr_vlog( RMR_VL_WARN, "cannot get epoll fd for rtg session; stats will generate only after update from rt manager\n" );
 485                 } else {
 486                         if( (epfd = epoll_create1( 0 )) < 0 ) {
 487                                 rmr_vlog( RMR_VL_WARN, "stats will generate only after rt manager update; unable to create epoll fd for rtg session: %s\n", strerror( errno ) );
 488                                 rcv_fd = -1;
 489                         } else {
 490                                 epe.events = EPOLLIN;
 491                                 epe.data.fd = rcv_fd;
 492
 493                                 if( epoll_ctl( epfd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 494                                         rmr_vlog( RMR_VL_WARN, "stats will generate only after rt manager update; unable to init epoll_ctl: %s\n", strerror( errno ) );
 495                                         rcv_fd = -1;
 496                                 }
 497                         }
 498                 }
 499         }
 500
 501         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", port );
 502         free( fport );
 503
 504         // future:  if we need to register with the rtg, then build a message and send it through a wormhole here
 505
 506         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 507         blabber = 0;
 508         while( 1 ) {                    // until the cows return, pigs fly, or somesuch event likely not to happen
 509                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 510                         if( rcv_fd < 0 || epoll_wait( epfd, events, 1, 1000 ) > 0 )  {  // skip epoll if init failed, else block for max 1 sec
 511                                 if( raw_interface ) {
 512                                         msg = (rmr_mbuf_t *) rcv_payload( pvt_cx, msg );                // receive from non-RMr sender
 513                                 } else {
 514                                         msg = rmr_rcv_msg( pvt_cx, msg );               // receive from an RMr sender
 515                                 }
 516                         } else {                                                                                                        // no msg, do extra tasks
 517                                 if( msg != NULL ) {                                                                             // if we were working with a message; ensure no len
 518                                         msg->len = 0;
 519                                         msg->state = RMR_ERR_TIMEOUT;
 520                                 }
 521                         }
 522
 523                         if( time( NULL ) > blabber  ) {
 524                                 vlevel = refresh_vlevel( vfd );
 525                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 526                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 527                                         if( blabber > bump_freq ) {
 528                                                 count_delay = 300;
 529                                         }
 530                                         rt_epcounts( ctx->rtable, ctx->my_name );
 531                                 }
 532                         }
 533                 }
 534
 535                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 536
 537                 if( msg != NULL && msg->len > 0 ) {
 538                         payload = msg->payload;
 539                         mlen = msg->len;                                        // usable bytes in the payload
 540                         if( vlevel > 1 ) {
 541                                 rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes (%s)\n", (int) mlen, msg->payload );
 542                         } else {
 543                                 if( DEBUG > 1 || (vlevel > 0) ) rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes\n", (int) mlen );
 544                         }
 545
 546                         if( pbuf_size <= mlen ) {
 547                                 if( pbuf ) {
 548                                         free( pbuf );
 549                                 }
 550                                 if( mlen < 512 ) {
 551                                         pbuf_size = 512;
 552                                 } else {
 553                                         pbuf_size = mlen * 2;
 554                                 }
 555                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 556                         }
 557                         memcpy( pbuf, payload, mlen );
 558                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 559
 560                         curr = pbuf;
 561                         while( curr ) {                                                         // loop over each record in the buffer
 562                                 nextr = strchr( curr, '\n' );                   // allow multiple newline records, find end of current and mark
 563
 564                                 if( nextr ) {
 565                                         *(nextr++) = 0;
 566                                 }
 567
 568                                 if( vlevel > 1 ) {
 569                                         rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: processing (%s)\n", curr );
 570                                 }
 571                                 if( raw_interface ) {
 572                                         parse_rt_rec( ctx, NULL, curr, vlevel, NULL );          // nil pvt to parser as we can't ack messages
 573                                 } else {
 574                                         parse_rt_rec( ctx, pvt_cx, curr, vlevel, msg );         // parse record and add to in progress table
 575                                 }
 576
 577                                 curr = nextr;
 578                         }
 579
 580                         if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 581                                 break;
 582                         }
 583
 584                         msg->len = 0;                           // force back into the listen loop
 585                 }
 586         }
 587
 588         return NULL;    // unreachable, but some compilers don't see that and complain.
 589 }
 590 #endif
 591
 592
 593 #endif