test/test_nng_em.c

   1 /*
   2 ==================================================================================
   3         Copyright (c) 2019 Nokia
   4         Copyright (c) 2018-2019 AT&T Intellectual Property.
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10            http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19
  20 /*
  21         Mnemonic:       test_nng_em.c
  22         Abstract:       A nano/NNG message emulator for testing without needing to
  23                                 actually have nanomsg, nng, or external processes.
  24                                 We also emulate the epoll_wait() function for controlled
  25                                 poll related testing.
  26
  27                                 This module must be directly included to be used.
  28         Date:           11 February 2019
  29         Author:         E. Scott Daniels
  30 */
  31
  32
  33 #include "rmr.h"                                // we use some of rmr defs in building dummy messages, so we need these
  34 #include "rmr_agnostic.h"
  35
  36 // ---------------------- emulated nng functions ---------------------------
  37
  38
  39 #ifndef _em_nn
  40 #define _em_nn
  41
  42 #include <pthread.h>
  43 #include "test_common_em.c"                     // things common to all emulation code
  44
  45 //--------------------------------------------------------------------------
  46 #ifdef EMULATE_NNG
  47 struct nn_msghdr {
  48         int boo;
  49 };
  50
  51 #define SOCKET_TYPE             nng_socket              // socket representation is different in each transport
  52
  53 /*
  54         Receive message must allocate a new buffer and return the pointer into *m.
  55         Every 9 messages or so we'll simulate an old version message
  56
  57         If em_mtc_msgs is set, then we add a non-zero d1 field with
  58         the call-id set to 2, and alternate the call flag
  59 */
  60 static int em_nng_recvmsg( nng_socket s, nng_msg ** m, int i ) {
  61         static int call_flag = 0;
  62
  63         void* b;
  64         struct em_msg* msg;
  65         int trace_size = 0;
  66         int d1_size = 0;
  67         unsigned char* d1;
  68
  69         if( rcv_delay > 0 ) {
  70                 sleep( rcv_delay );
  71         }
  72
  73         if( em_mtc_msgs ) {
  74                 d1_size = 4;
  75         }
  76
  77         if( m != NULL ) {
  78                 b = (void *) malloc( 2048 );
  79                 memset( b, 0, 2048 );
  80
  81                 *m = (nng_msg *) b;
  82                 msg = (struct em_msg *) b;
  83                 if( ! em_mtc_msgs  &&  (rcv_count % 10) == 9 ) {
  84                         msg->rmr_ver = ALT_MSG_VER;                                                     // allow emulation the bug in RMr v1
  85                 } else {
  86                         msg->rmr_ver = htonl( MSG_VER );
  87                 }
  88
  89                 msg->mtype = htonl( 1 );
  90                 msg->plen = htonl( 220 );
  91                 msg->len0 = htonl( sizeof( struct em_msg ) );
  92                 msg->len1 = htonl( trace_size );
  93                 msg->len2 = htonl( d1_size );
  94                 msg->len3 = htonl( 0 );
  95
  96                 pthread_mutex_lock( &rcv_gate );        // hold lock to update counter/flag
  97                 if( em_mtc_msgs ) {
  98                         d1 = DATA1_ADDR( msg );
  99                         d1[0] = 2;                                                                      // simulated msgs always on chute 2
 100                         if( call_flag ) {
 101                                 rcv_count++;
 102                                 msg->flags |= HFL_CALL_MSG;
 103                         }
 104                         if( rcv_delay > 0 ) {
 105                                 fprintf( stderr, "<EM>    count=%d flag=%d %02x \n", rcv_count, call_flag, msg->flags );
 106                         }
 107                         call_flag = !call_flag;
 108                 } else {
 109                         rcv_count++;
 110                 }
 111                 pthread_mutex_unlock( &rcv_gate );
 112                 snprintf( msg->xid, 32, "%015d", rcv_count );           // simple transaction id so we can test receive specific and ring stuff
 113                 snprintf( msg->src, 64, "localhost:4562" );             // set src id (unrealistic) so that rts() can be tested
 114                 snprintf( msg->srcip, 64, "89.2.19.19:4562" );          // set src ip for rts testing
 115
 116                 //fprintf( stderr, ">>> simulated received message: %s %s p=%p len0=%d\n", msg->src, msg->srcip, msg, (int) ntohl( msg->len0 ) );
 117         } else {
 118                 fprintf( stderr, "<WARN> em: simulated receive no msg pointer provided\n" );
 119         }
 120
 121         return return_value;
 122 }
 123
 124 static void* em_msg_body( nng_msg* msg ) {
 125         return (void *) msg;                                                            // we don't manage a real msg, so body is just the buffer we allocated
 126 }
 127
 128 static size_t em_msg_len( const nng_msg* msg ) {
 129         if( msg ) {
 130                 return  2048;
 131         }
 132
 133         return 0;
 134 }
 135
 136
 137 static int em_nng_pull_open(nng_socket * s ) {
 138         return return_value;
 139 }
 140 static int em_nng_pull0_open(nng_socket * s ) {
 141         return return_value;
 142 }
 143 static int em_nng_listen(nng_socket s, const char * c, nng_listener * l, int i ) {
 144         return return_value;
 145 }
 146 static int em_nng_close(nng_socket s ) {
 147         return return_value;
 148 }
 149 static int em_nng_push0_open(nng_socket * s ) {
 150         return return_value;
 151 }
 152 static int em_nng_dial(nng_socket s, const char * c, nng_dialer * d, int i ) {
 153         //fprintf( stderr, "<info> === simulated dialing: %s\n", c );
 154         return return_value;
 155 }
 156 static int em_nng_setopt(nng_socket s, const char * c, const void * p, size_t t ) {
 157         return return_value;
 158 }
 159 static int em_nng_sub_open(nng_socket * s ) {
 160         return return_value;
 161 }
 162 static int em_nng_sub0_open(nng_socket * s ) {
 163         return return_value;
 164 }
 165 static int em_nng_recv(nng_socket s, void * v, size_t * t, int i ) {
 166         return return_value;
 167 }
 168 static int em_nng_send( nng_socket s, void* m, int l, int f ) {
 169         free( m );                                      // we must ditch the message as nng does (or reuses)
 170         return return_value;
 171 }
 172
 173 /*
 174         Emulate sending a message. If the global em_send_failures is set,
 175         then every so often we fail with an EAGAIN to drive that part
 176         of the code in RMr.
 177 */
 178 static int em_sendmsg( nng_socket s, nng_msg* m, int i ) {
 179         static int count = 0;
 180
 181         if( em_send_failures && (count++ % 15 == 14) ) {
 182                 //fprintf( stderr, ">>>> failing send\n\n" );
 183                 return NNG_EAGAIN;
 184         }
 185
 186         return return_value;
 187 }
 188
 189 static void* em_nng_alloc( size_t len ) {
 190         return malloc( len );
 191 }
 192
 193 static int em_nng_msg_alloc( nng_msg** mp, size_t l ) {
 194         void*   p;
 195
 196         if( !mp || return_value != 0  ) {
 197                 return -1;
 198         }
 199
 200         p = (void *) malloc( sizeof( char ) * l );
 201         *mp = (nng_msg *) p;
 202
 203         return return_value;
 204 }
 205
 206 /*
 207         We just free the buffer here as it was a simple malloc.
 208 */
 209 static void em_nng_free( void* p, size_t l ) {
 210         if( p ) {
 211                 free( p );
 212         }
 213 }
 214 static void em_nng_msg_free( void* p ) {
 215         if( p ) {
 216                 free( p );
 217         }
 218 }
 219
 220 static int em_dialer_create( void* d, nng_socket s, char* stuff ) {
 221         //fprintf( stderr, ">>>> emulated dialer create\n\n" );
 222         return 0;
 223 }
 224
 225 static int em_dialer_start( nng_dialer d, int i ) {
 226         //fprintf( stderr, ">>>> emulated dialer start\n\n" );
 227         return return_value;
 228 }
 229
 230
 231 static int em_dialer_setopt_ms( nng_dialer dialer, void* option, int ms ) {
 232         return return_value;
 233 }
 234
 235 static int em_nng_getopt_int( nng_socket s, void* con, int* target ) {
 236         if( target ) {
 237                 *target = 0;
 238         }
 239         return return_value;
 240 }
 241
 242
 243
 244 // nng redefines some of these to point directly to various 'versions' of the function (ugg, function versions, really?)
 245 #undef nng_recvmsg
 246 #undef nng_free
 247 #undef nng_pull_open
 248 #undef nng_pull0_open
 249 #undef nng_listen
 250 #undef nng_close
 251 #undef nng_getopt_int
 252 #undef nng_push0_open
 253 #undef nng_dial
 254 #undef nng_setopt
 255 #undef nng_sub_open
 256 #undef nng_sub0_open
 257 #undef nng_recv
 258 #undef nng_alloc
 259
 260 #define nng_msg_alloc em_nng_msg_alloc
 261 #define nng_recvmsg em_nng_recvmsg
 262 #define nng_free em_nng_free
 263 #define nng_free em_nng_free
 264 #define nng_msg_free em_nng_msg_free
 265 #define nng_pull_open em_nng_pull_open
 266 #define nng_pull0_open em_nng_pull0_open
 267 #define nng_listen em_nng_listen
 268 #define nng_close em_nng_close
 269 #define nng_getopt_int em_nng_getopt_int
 270 #define nng_push0_open em_nng_push0_open
 271 #define nng_dial em_nng_dial
 272 #define nng_setopt em_nng_setopt
 273 #define nng_sub_open em_nng_sub_open
 274 #define nng_sub0_open em_nng_sub0_open
 275 #define nng_recv em_nng_recv
 276 #define nng_send em_nng_send
 277 #define nng_sendmsg em_sendmsg
 278 #define nng_alloc em_nng_alloc
 279 #define nng_free em_nng_free
 280 #define nng_dialer_setopt_ms em_dialer_setopt_ms
 281 #define nng_dialer_start em_dialer_start
 282 #define nng_dialer_create em_dialer_create
 283 #define nng_msg_body em_msg_body
 284 #define nng_msg_len em_msg_len
 285
 286
 287 #else
 288
 289 #define SOCKET_TYPE             int             // socket representation is different in each transport
 290
 291
 292 // ----------------------- emulated nano functions --------------------------
 293 struct em_nn_msghdr {
 294         int dummy;
 295 };
 296
 297 static int em_nn_socket (int domain, int protocol ) {
 298         static int s = 1;
 299
 300         return ++s;
 301 }
 302
 303 static int em_nn_close (int s ) {
 304         return 1;
 305 }
 306
 307 //static int em_nn_setsockopt (int s, int level, int option, const void *optval, size_t optvallen ) {
 308         //return 1;
 309 //}
 310
 311 static int em_nn_getsockopt (int s, int level, int option, void *optval, size_t *optvallen ) {
 312         return 1;
 313 }
 314
 315 static int em_nn_bind (int s, const char *addr ) {
 316         //      fprintf( stderr, ">>> ===== emulated bind called ====\n" );
 317         return 1;
 318 }
 319
 320 static int em_nn_connect (int s, const char *addr ) {
 321         return 1;
 322 }
 323
 324 static int em_nn_shutdown (int s, int how ) {
 325         return 1;
 326 }
 327
 328 static int em_nn_send (int s, const void *buf, size_t len, int flags ) {
 329         return 1;
 330 }
 331
 332 static int em_nn_recv (int s, void *m, size_t len, int flags ) {
 333         void* b;
 334         struct em_msg* msg;
 335         static int count = 0;                   // we'll simulate a message going in by dropping an rmr-ish msg with transaction id only
 336         int trace_size = 0;
 337         static int counter = 0;                         // if timeout value is set; we return timeout (eagain) every 3 calls
 338         int d1_size = 0;
 339
 340         if( em_timeout > 0 ) {
 341                 counter++;
 342                 if( counter % 3 == 0 ) {
 343                         return EAGAIN;
 344                 }
 345         }
 346
 347         if( em_mtc_msgs ) {
 348                 d1_size = 4;
 349         }
 350
 351         b = (void *) malloc( 2048 );
 352         if( m != NULL ) {                                               // blindly we assume this is 2k or bigger
 353                 memset( m, 0, 2048 );
 354                 msg = (struct em_msg *) m;
 355                 if( count % 10  == 9 ) {
 356                         //msg->rmr_ver = htonl( MSG_VER );
 357                         msg->rmr_ver = ALT_MSG_VER;             // emulate the bug in RMr v1
 358                 } else {
 359                         msg->rmr_ver = htonl( MSG_VER );
 360                 }
 361                 msg->mtype = htonl( 1 );
 362                 msg->plen = htonl( 220 );
 363                 msg->len0 = htonl( sizeof( struct em_msg ) );
 364                 msg->len1 = htonl( trace_size );
 365                 msg->len2 = htonl( d1_size );
 366                 msg->len3 = htonl( 0 );
 367                 snprintf( msg->xid, 32, "%015d", count++ );             // simple transaction id so we can test receive specific and ring stuff
 368                 snprintf( msg->src, 64, "localhost:4562" );             // set src id (unrealistic) so that rts() can be tested
 369                 snprintf( msg->srcip, 64, "89.2.19.19:4562" );          // set src ip for rts testing
 370                 //fprintf( stderr, "<EM>   returning message len=%d\n\n", ntohl( msg->plen ) );
 371         } else {
 372                 fprintf( stderr, "<EM>   message was nil\n\n" );
 373         }
 374
 375         //fprintf( stderr, ">>> simulated received message: %s %s len=%d p=%p\n", msg->src, msg->srcip, ntohl( msg->plen ), m );
 376         return 2048;
 377 }
 378
 379 static int em_sendmsg (int s, const struct em_nn_msghdr *msghdr, int flags ) {
 380         return 1;
 381 }
 382
 383 static int em_nn_recvmsg (int s, struct nn_msghdr *msghdr, int flags ) {
 384         return 1;
 385 }
 386
 387 static void em_nn_freemsg( void* ptr ) {
 388         free( ptr );
 389         return;
 390 }
 391
 392 /*
 393         Hacky implementation of set sock opt. We assume value is a pointer to int and ignore size.
 394 */
 395 static int em_setsockopt( int sock, int foo, int action, int* value, int size ) {
 396         if( action ==  NN_RCVTIMEO ) {
 397                 em_timeout = *value;
 398         }
 399 }
 400
 401
 402 // nanomsg
 403 #define nn_socket  em_nn_socket
 404 #define nn_close  em_nn_close
 405 //#define nn_setsockopt  em_nn_setsockopt
 406 #define nn_getsockopt  em_nn_getsockopt
 407 #define nn_bind  em_nn_bind
 408 #define nn_connect  em_nn_connect
 409 #define nn_shutdown  em_nn_shutdown
 410 #define nn_send  em_nn_send
 411 #define nn_recv  em_nn_recv
 412 #define nn_sendmsg  em_nn_sendmsg
 413 #define nn_recvmsg  em_nn_recvmsg
 414 #define nn_setsockopt  em_setsockopt
 415 #define nn_freemsg  em_nn_freemsg
 416
 417 #endif
 418
 419
 420 #endif