test/rtg_sim/rtm_sim.c

   1 // :vim ts=4 sw=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:       rtm_sim.c
  23         Abstract:       This is a simple route manager simulation which provides the ability
  24                                 to push a route table into one or more xAPPs. Designed just to
  25                                 drive the internal RMr route table collector outside of the static
  26                                 file allowing for testing of port definition in some containerised
  27                                 environments.
  28
  29                                 This application does not persist; it generates a set of tables based
  30                                 on the config file, connects to all applications listed, distributes
  31                                 the table, and exits.  If periodic delivery of one or more different
  32                                 configurations needs to be executed, use a shell script to wrap this
  33                                 application in a loop.
  34
  35         Date:           14 June 2019
  36         Author:         E. Scott Daniels
  37 */
  38
  39 /*
  40 config file format:
  41         #       comment and blank lines allowed
  42         #       trailing comments allowed
  43
  44         # port is used for any app listed in send2 which does not have a trailing :port
  45         # it may be supplied as a different value before each table, and if not
  46         # redefined applies to all subsequent tables.
  47         #
  48
  49         # A table consists of a send2 list (app[:port]) which are the applications that will
  50         # receive the table. Each table may contain one or more entries.  Entries define
  51         # the message type and subscription ID, along with one or more round robin groups.
  52         # A rrgroup is one or more app:port "endpoints" which RMr will use when sending
  53         # messages of the indicated type/subid.  Port on a rrgroup is rquired and is the
  54         # port that the application uses for app to app communications.
  55         #
  56
  57         port:   xapp-rtg-listen-port    # 4561 default
  58         table:
  59                 send2: app1:port app2:port ... appn:port
  60                 entry:
  61                         mtype:  n
  62                         subid:  n
  63                         rrgroup:        app:port... app:port
  64
  65                 entry:
  66                         mtype: n
  67                         subid: n
  68                         rrgroup: app:port ... app:port
  69 */
  70
  71
  72 #include <ctype.h>
  73 #include <unistd.h>
  74 #include <errno.h>
  75 #include <string.h>
  76 #include <stdio.h>
  77 #include <stdlib.h>
  78 #include <sys/epoll.h>
  79 #include  <time.h>
  80 #include <stdio.h>
  81 #include <fcntl.h>
  82 #include <sys/stat.h>
  83 #include <sys/types.h>
  84
  85 #include "req_resp.c"           // simple nano interface for request/response connections
  86
  87
  88 #define CONNECTED       1               // we've established a shunt connection to the app
  89
  90 #define TRUE            1
  91 #define FALSE           0
  92
  93 #define ALLOC_NEW       1               // rrsend should allocate a new buffer
  94
  95 #define MAX_TABLES      16              // total tables we support
  96 #define MAX_SEND2       64              // max number of apps that a table can be sent to
  97 #define MAX_APPS        1024    // max total apps defined (tables * send2)
  98 #define MAX_GROUPS      64              // max num of round robin groups per entry
  99 #define MAX_RRG_SIZE 64         // max number of apps in a group
 100 #define MAX_ENTRIES     256             // max entries in a table
 101 #define MAX_TOKENS      512             // max tokens we'll break a buffer into
 102
 103
 104 // ---------------------------------------------------------------------------------------
 105
 106 /*
 107         Things we need to track for an application.
 108 */
 109 typedef struct app {
 110         void*   shunt;                  // the rr context to shunt directly to an app
 111         int             state;                  // connected or not
 112         char*   name;                   // IP address or DNS name and port for connecting
 113         char*   port;                   // rr wants two strings as it builds it's own NN string
 114 } app_t;
 115
 116
 117
 118 // ----- table stuff (very staticly sized, but this isn't for prod) ----------------------
 119 /*
 120         A round robin group in a table entry
 121 */
 122 typedef struct rrgroup {
 123         int napps;                                              // number of apps
 124         char*   apps[MAX_RRG_SIZE];
 125 } rrgroup_t;
 126
 127 /*
 128         A single table entry.
 129 */
 130 typedef struct entry {
 131         int             mtype;                                  // entry message type
 132         int             subid;                                  // entry sub id
 133         int             ngroups;
 134         rrgroup_t       groups[MAX_GROUPS];     // the entry's groups
 135 } entry_t;
 136
 137 /*
 138         Defines a table which we will distribute.
 139 */
 140 typedef struct table {
 141         int             napps;                                          // number of apps this table is sent to
 142         int             first_app;                                      // first app in minfo that we send to
 143         int             nentries;                                       // number of entries
 144         entry_t entries[MAX_ENTRIES];
 145 } table_t;
 146
 147 /*
 148         Master set of contextual information.
 149 */
 150 typedef struct master {
 151         int             napps;                          // number in use (next insert point)
 152         int             ntables;
 153         int             port;                           // the port applications open by default for our connections
 154         app_t   apps[MAX_APPS];
 155         table_t tables[MAX_TABLES];
 156 } master_t;
 157
 158 /*
 159         Record buffer; file in memory which can be iterated over a record at a time.
 160 */
 161 typedef struct rbuffer {
 162         char*   buffer;                         // stuff read from file
 163         char*   rec;                            // next record
 164         int             at_end;                         // true if end was reached
 165 } rbuffer_t;
 166
 167 /*
 168         Set of tokens.
 169 */
 170 typedef struct tokens {
 171         char*   buffer;                                 // buffer that tokens points into
 172         int             ntoks;                                  // number of tokens in tokens
 173         char*   tokens[MAX_TOKENS];             // pointers into buffer at the start of each token 0..ntokens-1
 174 } tokens_t;
 175
 176
 177 // ----- token utilities ------------------------------------------------------------
 178
 179 /*
 180         Frees a token manager.
 181 */
 182 static void free_tokens( tokens_t* t ) {
 183         if( t == NULL ) {
 184                 return;
 185         }
 186
 187         free( t );
 188 }
 189
 190 /*
 191         Simple tokeniser.  If sep is whitespace, then leading whitespace (all, not just
 192         sep) is ignored; if sep is not whitespace, then leadign whitespace is included
 193         in the first token.  If sep is whitespace, consecutive instances of whitespace
 194         are treated as a single seperator:
 195                 if sep given as space, then
 196                 "bug boo" and "bug     boo"  both generate two tokens: (bug) (boo)
 197
 198                 if sep given as pipe (|), then
 199                 "bug||boo"   generates three tokens:  (bug), (), (boo)
 200
 201         Each token is a zero terminated string.
 202
 203 */
 204 static tokens_t* tokenise( char* buf, char sep ) {
 205         tokens_t*       t;
 206         int                     i;
 207         char            end_sep;                // if quoted endsep will be the quote mark
 208
 209         if( !buf || !(*buf) ) {
 210                 return NULL;
 211         }
 212
 213         t = (tokens_t *) malloc( sizeof( *t ) );
 214         memset( t, 0, sizeof( *t ) );
 215
 216         t->buffer = strdup( buf );
 217         buf = t->buffer;                                        // convenience
 218
 219         if( isspace( sep ) ) {                                                                          // if sep is in whitespace class
 220                 while( buf != NULL && *buf && isspace( *buf ) ) {               // pass over any leading whitespace
 221                         buf++;
 222                 }
 223
 224                 for( i = 0; i < strlen( buf ); i++ ) {
 225                         if( buf[i] == '\t' ) {
 226                                 buf[i] = ' ';
 227                         }
 228                 }
 229         }
 230
 231         while( buf != NULL && t->ntoks < MAX_TOKENS && *buf  ) {
 232                 if( *buf == '"' ) {
 233                         end_sep = '"';
 234                         buf++;
 235
 236                 } else {
 237                         end_sep = sep;
 238                 }
 239
 240                 t->tokens[t->ntoks++] = buf;                                            // capture token start
 241
 242                 if( (buf = strchr( buf, end_sep )) != NULL ) {                  // find token end
 243                         *(buf++) = 0;
 244
 245                         if( end_sep != sep ) {
 246                                 buf++;
 247                         }
 248
 249                         if( isspace( sep ) ) {                          // treat consec seperators as one if sep is whitespace
 250                                 while( *buf == sep ) {
 251                                         buf++;
 252                                 }
 253                         }
 254                 }
 255         }
 256
 257         return t;
 258 }
 259
 260 // ----- file/record management utilities ------------------------------------------------------------
 261
 262 /*
 263         Read an entire file into a single buffer.
 264 */
 265 static char* f2b( char* fname ) {
 266         struct stat     stats;
 267         off_t           fsize = 8192;   // size of the file
 268         off_t           nread;                  // number of bytes read
 269         int                     fd;
 270         char*           buf;                    // input buffer
 271
 272         if( (fd = open( fname, O_RDONLY )) >= 0 ) {
 273                 if( fstat( fd, &stats ) >= 0 ) {
 274                         if( stats.st_size <= 0 ) {                                      // empty file
 275                                 close( fd );
 276                                 fd = -1;
 277                         } else {
 278                                 fsize = stats.st_size;                                          // stat ok, save the file size
 279                         }
 280                 } else {
 281                         fsize = 8192;                                                           // stat failed, we'll leave the file open and try to read a default max of 8k
 282                 }
 283         }
 284
 285         if( fd < 0 ) {                                                                                  // didn't open or empty
 286                 if( (buf = (char *) malloc( sizeof( char ) * 1 )) == NULL ) {
 287                         return NULL;
 288                 }
 289
 290                 *buf = 0;
 291                 return buf;
 292         }
 293
 294         if( (buf = (char *) malloc( sizeof( char ) * fsize + 2 )) == NULL ) {           // enough to add nil char to make string
 295                 close( fd );
 296                 errno = ENOMEM;
 297                 return NULL;
 298         }
 299
 300         nread = read( fd, buf, fsize );
 301         if( nread < 0 || nread > fsize ) {                                                      // failure of some kind
 302                 free( buf );
 303                 errno = EFBIG;                                                                                  // likely too much to handle
 304                 close( fd );
 305                 return NULL;
 306         }
 307
 308         buf[nread] = 0;
 309
 310         close( fd );
 311         return buf;
 312 }
 313
 314 /*
 315         Read a file into a buffer, and set a record buffer to manage it.
 316 */
 317 static rbuffer_t* f2r( char* fname ) {
 318         char*           raw;                    // raw buffer
 319         rbuffer_t*      r;
 320
 321         if( (raw = f2b( fname )) == NULL ) {
 322                 return NULL;
 323         }
 324
 325         r = (rbuffer_t *) malloc( sizeof( *r ) );
 326         memset( r, 0, sizeof( *r ) );
 327         r->buffer = raw;
 328         r->rec = raw;                                                   // point at first (only) record
 329 }
 330
 331 /*
 332         Return a pointer to the next record in the buffer, or nil if at
 333         end of buffer.
 334 */
 335 static char* next_rec( rbuffer_t* r ) {
 336         char*   rec;
 337
 338         if( !r || r->at_end ) {
 339                 return NULL;
 340         }
 341
 342         rec = r->rec;
 343         r->rec = strchr( r->rec, '\n' );
 344         if( r->rec ) {
 345                 *r->rec = 0;
 346                 r->rec++;
 347                 if( *r->rec == 0 ) {
 348                         r->at_end = TRUE;
 349                 }
 350         } else {
 351                 r->at_end = TRUE;                               // mark for next call
 352         }
 353
 354         return rec;
 355 }
 356
 357 static void free_rbuf( rbuffer_t* r ) {
 358         if( r != NULL ) {
 359                 if( r->buffer != NULL ) {
 360                         free( r->buffer );
 361                 }
 362
 363                 free( r );
 364         }
 365 }
 366
 367
 368 // ----- table management ------------------------------------------------------------
 369
 370 /*
 371         Run the app list and attempt to open a shunt to any unconnected application.
 372         Returns the number of applications that could not be connected to.
 373 */
 374 static int connect2all( master_t* mi ) {
 375         int     errors = 0;
 376         int i;
 377
 378
 379         if( mi == NULL ) {
 380                 return 1;
 381         }
 382
 383         for( i = 0; i < mi->napps; i++ ) {
 384                 if( mi->apps[i].state != CONNECTED ) {
 385                         fprintf( stderr, "[INF] opening shunt to: %s:%s\n", mi->apps[i].name, mi->apps[i].port );
 386                         mi->apps[i].shunt = rr_connect( mi->apps[i].name, mi->apps[i].port );
 387                         if( mi->apps[i].shunt == NULL) {
 388                                 errors++;
 389                         } else {
 390                                 fprintf( stderr, "[INFO] shunt created to: %s:%s\n", mi->apps[i].name, mi->apps[i].port );
 391                                 mi->apps[i].state = CONNECTED;
 392                         }
 393                 }
 394         }
 395
 396         return errors;
 397 }
 398
 399 /*
 400         Add an application to the current table.
 401 */
 402 static void add_app( master_t* mi, char* app_name ) {
 403         char    wbuf[256];
 404         char*   app_port;
 405         char*   ch;
 406
 407         if( mi == NULL || app_name == NULL ) {
 408                 return;
 409         }
 410
 411         if( mi->napps > MAX_APPS ) {
 412                 fprintf( stderr, "[WARN] too many applications, ignoring: %s\n", app_name );
 413         }
 414
 415         if( (ch = strchr( app_name, ':' )) == NULL ) {          // assume we are using the default rm listen port
 416                 snprintf( wbuf, sizeof( wbuf ), "%d", mi->port );
 417                 app_port = wbuf;
 418         } else {
 419                 *(ch++) = 0;                                                                    // name port given, split and point at port
 420                 app_port = ch;
 421         }
 422
 423         mi->apps[mi->napps].name = strdup( app_name );
 424         mi->apps[mi->napps].port = strdup( app_port );
 425         mi->apps[mi->napps].state = !CONNECTED;
 426         mi->napps++;
 427 }
 428
 429
 430 /*
 431         Initialise things; returns a master info context.
 432 */
 433 static master_t* init( char* cfname ) {
 434         master_t*       mi;
 435         char            wbuf[128];
 436         rbuffer_t*      rb;                                                     // record manager for reading config file
 437         char*           rec;
 438         tokens_t*       tokens;
 439         int                     i;
 440         int                     errors;
 441         char*           tok;
 442         table_t*        table = NULL;
 443         rrgroup_t*      rrg = NULL;
 444         entry_t*        entry = NULL;
 445         int                     rec_num = 0;
 446
 447         mi = (master_t *) malloc( sizeof( *mi ) );
 448         if( mi == NULL ) {
 449                 return NULL;
 450         }
 451         memset( mi, 0, sizeof( *mi ) );
 452         mi->port = 4561;
 453
 454         rb = f2r( cfname );                                             // get a record buffer to parse the config file
 455         if( rb == NULL ) {
 456                 fprintf( stderr, "[FAIL] unable to open config file: %s: %s\n", cfname, strerror( errno ) );
 457                 free( mi );
 458                 return NULL;
 459         }
 460
 461         while( (rec = next_rec( rb )) != NULL ) {
 462                 if( *rec ) {
 463                         rec_num++;
 464                         tokens = tokenise( rec, ' ' );
 465
 466                         fprintf( stderr, "parsing %d: %s\n", rec_num, rec );
 467
 468                         for( i = 0; i < tokens->ntoks && *tokens->tokens[i] != '#'; i++ );                      // simple comment strip
 469                         tokens->ntoks = i;
 470
 471                         if( tokens->ntoks > 0 ) {
 472                                 tok = tokens->tokens[0];
 473                                 switch( *tok ) {                                        // faster jump table based on 1st ch; strcmp only if needed later
 474                                         case '#':
 475                                                 break;
 476
 477                                         case 'e':
 478                                                 if( table != NULL  && table->nentries < MAX_ENTRIES ) {
 479                                                         entry = &table->entries[table->nentries++];
 480                                                         entry->subid = -1;                      // no subscription id if user omits
 481                                                 } else {
 482                                                         fprintf( stderr, "[ERR] @%d no table started, or table full\n", rec_num );
 483                                                 }
 484                                                 break;
 485
 486                                         case 'm':
 487                                                 if( entry != NULL ) {
 488                                                         entry->mtype = atoi( tokens->tokens[1] );
 489                                                 } else {
 490                                                         fprintf( stderr, "[ERR] @%d no entry started\n", rec_num );
 491                                                 }
 492                                                 break;
 493
 494                                         case 'p':
 495                                                 if( tokens->ntoks > 1 ) {
 496                                                         mi->port = atoi( tokens->tokens[1] );
 497                                                         if( mi->port < 1000 ) {
 498                                                                 fprintf( stderr, "[WRN] @%d assigned default xAPP port smells fishy: %s\n", rec_num, tokens->tokens[1] );
 499                                                         }
 500                                                 }
 501                                                 break;
 502
 503                                         case 'r':                                                       // round robin group
 504                                                 if( entry != NULL && entry->ngroups < MAX_GROUPS ) {
 505                                                         if( tokens->ntoks < MAX_RRG_SIZE ) {
 506                                                                 rrg = &entry->groups[entry->ngroups++];
 507
 508                                                                 for( i = 1; i < tokens->ntoks; i++ ) {
 509                                                                         rrg->apps[rrg->napps++] = strdup( tokens->tokens[i] );
 510                                                                 }
 511                                                         } else {
 512                                                                 fprintf( stderr, "[ERR] @%d round robin group too big.\n", rec_num );
 513                                                         }
 514                                                 } else {
 515                                                         fprintf( stderr, "[ERR] @%d no previous entry, or entry is full\n", rec_num );
 516                                                 }
 517                                                 break;
 518
 519                                         case 's':
 520                                                 if( *(tok+1) == 'e' ) {                 // send2
 521                                                         if( table != NULL && tokens->ntoks < MAX_SEND2  ) {
 522                                                                 table->first_app = mi->napps;
 523
 524                                                                 for( i = 1; i < tokens->ntoks; i++ ) {
 525                                                                         add_app( mi, tokens->tokens[i] );
 526                                                                 }
 527
 528                                                                 table->napps = tokens->ntoks - 1;
 529                                                         }
 530                                                 } else {                                                // subid
 531                                                         if( entry != NULL ) {
 532                                                                 entry->subid = atoi( tokens->tokens[1] );
 533                                                         } else {
 534                                                                 fprintf( stderr, "[ERR] @%d no entry started\n", rec_num );
 535                                                         }
 536                                                 }
 537                                                 break;
 538
 539                                         case 't':
 540                                                 entry = NULL;
 541                                                 if( mi->ntables < MAX_TABLES ) {
 542                                                         table = &mi->tables[mi->ntables++];
 543                                                 } else {
 544                                                         fprintf( stderr, "[ERR] @%d too many tables defined\n", rec_num );
 545                                                         table = NULL;
 546                                                 }
 547                                                 break;
 548
 549                                         default:
 550                                                 fprintf( stderr, "record from config was ignored: %s\n", rec );
 551                                                 break;
 552                                 }
 553                         }
 554                 }
 555         }
 556
 557         free_rbuf( rb );
 558         return mi;
 559 }
 560
 561
 562 /*
 563         Build a buffer with the entry n from table t. Both entry and table
 564         numbers are 0 based.
 565         Caller must free returned buffer.
 566 */
 567 static char* mk_entry( master_t* mi, int table, int entry ) {
 568         char            wbuf[4096];
 569         char            sbuf[256];
 570         table_t*        tab;
 571         entry_t*        ent;
 572         int                     i;
 573         int                     j;
 574         int                     len;
 575         int                     alen;
 576
 577         if( !mi || mi->ntables <= table ) {
 578                 return NULL;
 579         }
 580
 581         tab = &mi->tables[table];
 582         if( tab->nentries <= entry ) {
 583                 return NULL;
 584         }
 585
 586         ent = &tab->entries[entry];
 587
 588         snprintf( wbuf, sizeof( wbuf ), "mse | %d | %d | ", ent->mtype, ent->subid  );                  // we only generate mse records
 589
 590         len = strlen( wbuf );
 591         for( i = 0; i < ent->ngroups; i++ ) {
 592                 if( i ) {
 593                         strcat( wbuf, "; " );
 594                 }
 595
 596                 for( j = 0; j < ent->groups[i].napps; j++ ) {
 597                         alen = strlen( ent->groups[i].apps[j] ) + 3;                            // not percise, but close enough for testing
 598                         if( alen + len > sizeof( wbuf ) ) {
 599                                 fprintf( stderr, "[ERR] entry %d for table %d is too large to format\n", entry, table );
 600                                 return NULL;
 601                         }
 602
 603                         if( j ) {
 604                                 strcat( wbuf, "," );
 605                         }
 606
 607                         strcat( wbuf, ent->groups[i].apps[j] );
 608                 }
 609         }
 610
 611         strcat( wbuf, "\n" );
 612         return strdup( wbuf );
 613 }
 614
 615 /*
 616         Sends a buffer to all apps in the range.
 617 */
 618 void send2range( master_t* mi, char* buf, int first, int n2send ) {
 619         int                     a;                                              // application offset in master array
 620         int                     last;                                   // stopping point (index)
 621         rr_mbuf_t*      mbuf = NULL;
 622
 623         if( !mi  ) {                                                                                    // safe to dance
 624                 return;
 625         }
 626
 627         a = first;
 628         last = first + n2send;
 629
 630         mbuf = rr_new_buffer( mbuf, strlen( buf ) + 5 );                // ensure buffer is large enough
 631         while(  a < last ) {
 632                 fprintf( stderr, "%s ", mi->apps[a].name );
 633
 634                 memcpy( mbuf->payload, buf, strlen( buf ) );
 635                 mbuf->used = strlen( buf );
 636                 mbuf = rr_send( mi->apps[a].shunt, mbuf, ALLOC_NEW );
 637
 638                 a++;
 639
 640                 fprintf( stderr, "\n" );
 641         }
 642
 643         rr_free_mbuf( mbuf );
 644 }
 645
 646 /*
 647         Formats the entries for the table and sends to all applications that the
 648         table should be sent to per the send2 directive in the config.
 649 */
 650 static void send_table( master_t* mi, int table ) {
 651         table_t*        tab;
 652         char*           ent;                            // entry string to send
 653         int                     e = 0;                          // entry number
 654         int                     a;
 655         int                     last;
 656         rr_mbuf_t*      mbuf = NULL;
 657
 658         if( !mi || table > mi->ntables ) {
 659                 return;
 660         }
 661
 662         tab = &mi->tables[table];
 663         fprintf( stderr, "[INF] send table start message: " );
 664         send2range( mi, "newrt | start\n", tab->first_app, tab->napps );                // send table start req to all
 665
 666         while( (ent = mk_entry( mi, table, e++ )) != NULL ) {                                   // build each entry once, send to all
 667                 fprintf( stderr, "[INF] sending table %d entry %d: ", table, e );
 668                 send2range( mi, ent, tab->first_app, tab->napps );
 669         }
 670
 671         fprintf( stderr, "[INF] send table end message: " );
 672         send2range( mi, "newrt | end\n", tab->first_app, tab->napps );          // send table end notice to all
 673 }
 674
 675 /*
 676         Run the list of tables and send them all out.
 677 */
 678 static void send_all_tables( master_t* mi ) {
 679         int i;
 680
 681         if( ! mi ) {
 682                 return;
 683         }
 684
 685         for( i = 0; i < mi->ntables; i++ ) {
 686                 send_table( mi, i );
 687         }
 688 }
 689
 690
 691 // ----------------- testing ----------------------------------------------------------------------
 692
 693 /*
 694         Dump table entries in the form we'll send to apps to stderr.
 695 */
 696 static void print_tables( master_t* mi ) {
 697         char*   ent;
 698         int             t;
 699         int             e;
 700
 701         if( ! mi ) {
 702                 return;
 703         }
 704
 705         for( t = 0; t < mi->ntables; t++ ) {
 706                 fprintf( stderr, "=== table %d ===\n", t );
 707                 e = 0;
 708                 while( (ent = mk_entry( mi, t, e++ )) != NULL ) {
 709                         fprintf( stderr, "%s", ent );
 710                         free( ent );
 711                 }
 712         }
 713 }
 714
 715 static void print_tokens( tokens_t* tokens ) {
 716         int i;
 717
 718         fprintf( stderr, "there are %d tokens\n", tokens->ntoks );
 719         for( i = 0; i < tokens->ntoks; i++ ) {
 720                 fprintf( stderr, "[%02d] (%s)\n", i, tokens->tokens[i] );
 721         }
 722 }
 723
 724 static void self_test( char* fname ) {
 725         char*   s0 = "            Now is   the time for     all    to stand up and cheer!";
 726         char*   s1 = "            Now is   \"the time for     all\"    to stand up and cheer!";
 727         char*   s2 = " field1 | field2||field4|field5";
 728         tokens_t*       tokens;
 729         rbuffer_t       *rb;
 730         int i;
 731
 732         tokens = tokenise( s0, ' ' );
 733         if( tokens->ntoks != 11 ) {
 734                 fprintf( stderr, "didn't parse into 11 tokens (got %d): %s\n", tokens->ntoks, s1 );
 735         }
 736         print_tokens( tokens );
 737
 738         tokens = tokenise( s1, ' ' );
 739         if( tokens->ntoks != 8 ) {
 740                 fprintf( stderr, "didn't parse into 11 tokens (got %d): %s\n", tokens->ntoks, s1 );
 741         }
 742         print_tokens( tokens );
 743
 744         tokens = tokenise( s2, '|' );
 745         if( tokens->ntoks != 5 ) {
 746                 fprintf( stderr, "didn't parse into 5 tokens (got %d): %s\n", tokens->ntoks, s2 );
 747         }
 748         print_tokens( tokens );
 749
 750         free_tokens( tokens );
 751
 752         if( fname == NULL ) {
 753                 return;
 754         }
 755
 756         rb = f2r( fname );
 757         if( rb == NULL ) {
 758                 fprintf( stderr, "[FAIL] couldn't read file into rbuffer: %s\n", strerror( errno ) );
 759         } else {
 760                 while( (s2 = next_rec( rb )) != NULL ) {
 761                         fprintf( stderr, "record: (%s)\n", s2 );
 762                 }
 763
 764                 free_rbuf( rb );
 765         }
 766
 767         return;
 768 }
 769
 770
 771 // ----------------------------------------------------------------------------------------------
 772
 773 int main( int argc, char** argv ) {
 774         void*   mi;
 775         int             not_ready;
 776
 777         if( argc > 1 ) {
 778                 if( strcmp( argv[1], "selftest" ) == 0 ) {
 779                         self_test( argc > 1 ? argv[2] : NULL );
 780                         exit( 0 );
 781                 }
 782
 783                 mi = init( argv[1] );                                                                   // parse the config and generate table structs
 784                 if( ! mi ) {
 785                         fprintf( stderr, "[CRI] initialisation failed\n" );
 786                         exit( 1 );
 787                 }
 788
 789                 print_tables( mi );
 790
 791                 while( (not_ready = connect2all( mi ) ) > 0 ) {
 792                         fprintf( stderr, "[INF] still waiting to connect to %d applications\n", not_ready );
 793                         sleep( 2 );
 794                 }
 795
 796                 fprintf( stderr, "[INF] connected to all applications, sending tables\n" );
 797
 798                 send_all_tables( mi );
 799         } else {
 800                 fprintf( stderr, "[INFO] usage: %s file-name\n", argv[0] );
 801         }
 802 }
 803