examples/sender.c

   1 // :vim ts=4 sw=4 noet:
   2
   3 /*
   4         Mnemonic:       sender.c
   5         Abstract:       Very simple test sender. Sends messages with a given delay between each.
   6                                 The sender also uses epoll_wait() to ensure that any received messages
   7                                 don't clog the queue.
   8
   9         Parms:          The following positional parameters are recognised on the command line:
  10                                         [listen_port [delay [stats-freq] [msg-type]]]]
  11
  12                                         Defaults:
  13                                                 listen_port 43086
  14                                                 delay (mu-sec) 1000000 (1 sec)
  15                                                 stats-freq 10
  16                                                 msg-type 0
  17
  18         Date:           1 April 2019
  19 */
  20
  21 #include <unistd.h>
  22 #include <errno.h>
  23 #include <string.h>
  24 #include <stdio.h>
  25 #include <stdlib.h>
  26 #include <sys/epoll.h>
  27 #include <time.h>
  28
  29 #include <rmr/rmr.h>
  30
  31 int main( int argc, char** argv ) {
  32     void* mrc;                                                  //msg router context
  33     struct epoll_event events[1];                       // list of events to give to epoll
  34     struct epoll_event epe;                 // event definition for event to listen to
  35     int     ep_fd = -1;                                         // epoll's file des (given to epoll_wait)
  36     int rcv_fd;                                                 // file des that NNG tickles -- give this to epoll to listen on
  37         int nready;                                                             // number of events ready for receive
  38         rmr_mbuf_t*             sbuf;                                   // send buffer
  39         rmr_mbuf_t*             rbuf;                                   // received buffer
  40         int     count = 0;
  41         int     rcvd_count = 0;
  42         char*   listen_port = "43086";
  43         int             delay = 1000000;                                                // mu-sec delay between messages
  44         int             mtype = 0;
  45         int             stats_freq = 100;
  46
  47         if( argc > 1 ) {
  48                 listen_port = argv[1];
  49         }
  50         if( argc > 2 ) {
  51                 delay = atoi( argv[2] );
  52         }
  53         if( argc > 3 ) {
  54                 mtype = atoi( argv[3] );
  55         }
  56
  57         fprintf( stderr, "<DEMO> listen port: %s; mtype: %d; delay: %d\n", listen_port, mtype, delay );
  58
  59     if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
  60                 fprintf( stderr, "<DEMO> unable to initialise RMr\n" );
  61                 exit( 1 );
  62         }
  63
  64     rcv_fd = rmr_get_rcvfd( mrc );                                      // set up epoll things, start by getting the FD from MRr
  65         if( rcv_fd < 0 ) {
  66                 fprintf( stderr, "<DEMO> unable to set up polling fd\n" );
  67                 exit( 1 );
  68         }
  69         if( (ep_fd = epoll_create1( 0 )) < 0 ) {
  70                 fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
  71                 exit( 1 );
  72         }
  73     epe.events = EPOLLIN;
  74     epe.data.fd = rcv_fd;
  75
  76     if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
  77                 fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
  78                 exit( 1 );
  79         }
  80
  81         sbuf = rmr_alloc_msg( mrc, 256 );       // alloc first send buffer; subsequent buffers allcoated on send
  82         rbuf = NULL;                                            // don't need to alloc receive buffer
  83
  84         while( ! rmr_ready( mrc ) ) {           // must have a route table before we can send; wait til RMr say it has one
  85                 sleep( 1 );
  86         }
  87         fprintf( stderr, "<DEMO> rmr is ready\n" );
  88
  89
  90     while( 1 ) {                                                                                // send messages until the cows come home
  91                 snprintf( sbuf->payload, 200, "count=%d received= %d ts=%lld %d stand up and cheer!",   // create the payload
  92                         count, rcvd_count, (long long) time( NULL ), rand() );
  93
  94                 sbuf->mtype = mtype;                                                    // fill in the message bits
  95                 sbuf->len =  strlen( sbuf->payload ) + 1;               // our receiver likely wants a nice acsii-z string
  96                 sbuf->state = 0;
  97                 sbuf = rmr_send_msg( mrc, sbuf );                               // send it (send returns an empty payload on success, or the original payload on fail/retry)
  98                 while( sbuf->state == RMR_ERR_RETRY ) {                 // soft failure (device busy?) retry
  99                         sbuf = rmr_send_msg( mrc, sbuf );                       // retry send until it's good (simple test; real programmes should do better)
 100                 }
 101                 count++;
 102
 103                 while( (nready = epoll_wait( ep_fd, events, 1, 0 )) > 0 ) {     // if something ready to receive (non-blocking check)
 104                         if( events[0].data.fd == rcv_fd ) {             // we only are waiting on 1 thing, so [0] is ok
 105                                 errno = 0;
 106                                 rbuf = rmr_rcv_msg( mrc, rbuf );
 107                                 if( rbuf ) {
 108                                         rcvd_count++;
 109                                 }
 110                         }
 111                 }
 112
 113                 if( (count % stats_freq) == 0 ) {
 114                         fprintf( stderr, "<DEMO> sent %d   received %d\n", count, rcvd_count );
 115                 }
 116
 117                 usleep( delay );
 118     }
 119 }
 120