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