examples/rmr_dump.cpp

   1 // vi: ts=4 sw=4 noet:
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2020 Nokia
   5         Copyright (c) 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:       rmr_dump.cpp
  23         Abstract:       This is a sample xapp which will behave somewhat like tcpdump
  24                                 as it receives rmr messages.  The list of message types which should
  25                                 be processed are given on the command line, following any flag
  26                                 options (e.g. -p port).  Any of those messages received are reported
  27                                 on by the application.  The verbosity level may be used to increase
  28                                 the amount of detail given for the tracked messages.
  29
  30                                 This writes to the TTY which is slow, so do not expect it to be able
  31                                 to process and report on a high rate of messages. Also, forwarded
  32                                 messages will reach the intended target, however if the target
  33                                 attempts to send a response the response will come back to THIS
  34                                 application, and not the message origination; this cannot be a bit
  35                                 of middleware in it's current form.
  36
  37         Date:           25 March 2020
  38         Author:         E. Scott Daniels
  39
  40 */
  41 #include <stdio.h>
  42 #include <unistd.h>
  43 #include <atomic>
  44
  45 #include "ricxfcpp/xapp.hpp"
  46
  47 /*
  48         Information that the callback needs outside
  49         of what is given to it via parms on a call
  50         by the framework.
  51 */
  52 typedef struct {
  53         int             vlevel;                         // verbosity level
  54         bool    forward;                        // if true, message is forwarded
  55         int             stats_freq;                     // header/stats after n messages
  56         std::atomic<long>       pcount; // messages processed
  57         std::atomic<long>       icount; // messages ignored
  58         std::atomic<int>        hdr;    // number of messages before next header
  59 } cb_info_t;
  60
  61 // ----------------------------------------------------------
  62
  63 /*
  64         Dump bytes to tty.
  65 */
  66 void dump( unsigned const char* buf, int len ) {
  67         int             i;
  68         int             j;
  69         char    cheater[17];
  70
  71         fprintf( stdout, "<RD> 0000 | " );
  72         j = 0;
  73         for( i = 0; i < len; i++ ) {
  74                 cheater[j++] =  isprint( buf[i] ) ? buf[i] : '.';
  75                 fprintf( stdout, "%02x ", buf[i] );
  76
  77                 if( j == 16 ) {
  78                         cheater[j] = 0;
  79                         fprintf( stdout, " | %s\n<RD> %04x | ", cheater, i+1 );
  80                         j = 0;
  81                 }
  82         }
  83
  84         if( j ) {
  85                 i = 16 - (i % 16);
  86                 for( ; i > 0; i-- ) {
  87                         fprintf( stdout, "   " );
  88                 }
  89                 cheater[j] = 0;
  90                 fprintf( stdout, " | %s\n", cheater );
  91         }
  92 }
  93
  94 /*
  95         generate stats when the hdr count reaches 0. Only one active
  96         thread will ever see it be exactly 0, so this is thread safe.
  97 */
  98 void stats( cb_info_t& cbi ) {
  99         int curv;                                       // current stat trigger value
 100
 101         curv = cbi.hdr--;
 102
 103         if( curv == 0 ) {                                       // stats when we reach 0
 104                 fprintf( stdout, "ignored: %ld  processed: %ld\n",
 105                         cbi.icount.load(), cbi.pcount.load() );
 106                 if( cbi.vlevel > 0 ) {
 107                         fprintf( stdout, "\n     %5s %5s %2s %5s\n",
 108                                 "MTYPE", "SUBID", "ST", "PLLEN" );
 109                 }
 110
 111                 cbi.hdr = cbi.stats_freq;               // reset must be last
 112         }
 113 }
 114
 115 void cb1( Message& mbuf, int mtype, int subid, int len,
 116                                 Msg_component payload,  void* data ) {
 117         cb_info_t*      cbi;
 118         long total_count;
 119
 120         if( (cbi = (cb_info_t *) data) == NULL ) {
 121                 return;
 122         }
 123
 124         cbi->pcount++;
 125         stats( *cbi );                  // gen stats & header if needed
 126
 127         if( cbi->vlevel > 0 ) {
 128                 fprintf( stdout, "<RD> %-5d %-5d %02d %-5d \n",
 129                                 mtype, subid, mbuf.Get_state(), len );
 130
 131                 if( cbi->vlevel > 1 ) {
 132                         dump(  payload.get(), len > 64 ? 64 : len );
 133                 }
 134         }
 135
 136         if( cbi->forward ) {
 137                 // forward with no change to len or payload
 138                 mbuf.Send_msg( Message::NO_CHANGE, NULL );
 139         }
 140 }
 141
 142 /*
 143         registered as the default callback; it counts the
 144         messages that we aren't giving details about.
 145 */
 146 void cbd( Message& mbuf, int mtype, int subid, int len,
 147                                 Msg_component payload,  void* data ) {
 148         cb_info_t*      cbi;
 149
 150         if( (cbi = (cb_info_t *) data) == NULL ) {
 151                 return;
 152         }
 153
 154         cbi->icount++;
 155         stats( *cbi );
 156
 157         if( cbi->forward ) {
 158                 // forward with no change to len or payload
 159                 mbuf.Send_msg( Message::NO_CHANGE, NULL );
 160         }
 161 }
 162
 163 int main( int argc, char** argv ) {
 164         std::unique_ptr<Xapp> x;
 165         char*   port = (char *) "4560";
 166         int ai = 1;                                     // arg processing index
 167         cb_info_t*      cbi;
 168         int             ncb = 0;                        // number of callbacks registered
 169         int             mtype;
 170         int             nthreads = 1;
 171
 172         cbi = (cb_info_t *) malloc( sizeof( *cbi ) );
 173         cbi->pcount = 0;
 174         cbi->icount = 0;
 175         cbi->stats_freq = 10;
 176
 177         ai = 1;
 178         // very simple flag parsing (no error/bounds checking)
 179         while( ai < argc ) {
 180                 if( argv[ai][0] != '-' )  {             // break on first non-flag
 181                         break;
 182                 }
 183
 184                 // very simple arg parsing; each must be separate -x -y not -xy.
 185                 switch( argv[ai][1] ) {
 186                         case 'f':                                       // enable packet forwarding
 187                                 cbi->forward = true;
 188                                 break;
 189
 190                         case 'p':                                       // define port
 191                                 port = argv[ai+1];
 192                                 ai++;
 193                                 break;
 194
 195                         case 's':                                               // stats frequency
 196                                 cbi->stats_freq = atoi( argv[ai+1] );
 197                                 if( cbi->stats_freq < 5 ) {     // enforce sanity
 198                                         cbi->stats_freq = 5;
 199                                 }
 200                                 ai++;
 201                                 break;
 202
 203                         case 't':                                               // thread count
 204                                 nthreads = atoi( argv[ai+1] );
 205                                 if( nthreads < 1 ) {
 206                                         nthreads = 1;
 207                                 }
 208                                 ai++;
 209                                 break;
 210
 211                         case 'v':                       // simple verbose bump
 212                                 cbi->vlevel++;
 213                                 break;
 214
 215                         case 'V':                       // explicit verbose level
 216                                 cbi->vlevel = atoi( argv[ai+1] );
 217                                 ai++;
 218                                 break;
 219
 220                         default:
 221                                 fprintf( stderr, "unrecognised option: %s\n", argv[ai] );
 222                                 fprintf( stderr, "usage: %s [-f] [-p port] [-s stats-freq]  [-t thread-count] [-v | -V n] msg-type1 ... msg-typen\n", argv[0] );
 223                                 fprintf( stderr, "\tstats frequency is in number of messages received\n" );
 224                                 fprintf( stderr, "\tverbose levels (-V) 0 counts only, 1 message info 2 payload dump\n" );
 225                                 exit( 1 );
 226                 }
 227
 228                 ai++;
 229         }
 230
 231         cbi->hdr = cbi->stats_freq;
 232         fprintf( stderr, "<RD> listening on port: %s\n", port );
 233
 234         // create xapp, wait for route table if forwarding
 235         x = std::unique_ptr<Xapp>( new Xapp( port, cbi->forward ) );
 236
 237         // register callback for each type on the command line
 238         while( ai < argc ) {
 239                 mtype = atoi( argv[ai] );
 240                 ai++;
 241                 fprintf( stderr, "<RD> capturing messages for type %d\n", mtype );
 242                 x->Add_msg_cb( mtype, cb1, cbi );
 243                 ncb++;
 244         }
 245
 246         if( ncb < 1 ) {
 247                 fprintf( stderr, "<RD> no message types specified on the command line\n" );
 248                 exit( 1 );
 249         }
 250
 251         x->Add_msg_cb( x->DEFAULT_CALLBACK, cbd, cbi );         // register default cb
 252
 253         fprintf( stderr, "<RD> starting driver\n" );
 254         x->Run( nthreads );
 255
 256         // return from run() is not expected, but some compilers might
 257         // compilain if there isn't a return value here.
 258         return 0;
 259 }