e2sim/e2apv1sim/test/rmr_interface/tests/sender/rmr_wrapper.h

   1 /*
   2  *
   3  * Copyright 2019 AT&T Intellectual Property
   4  * Copyright 2019 Nokia
   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 int epoll_to = 1;// global default epoll timout -- 1ms
  21 char*   eparm;                                                  // generic env pointer
  22
  23 struct rmr_context {
  24         void* mrc;
  25         int rcv_fd;     // pollable fd
  26         struct epoll_event epe;                 // event definition for event to listen to
  27         struct epoll_event events[10];          // wait on 10 possible events
  28         int     ep_fd;
  29         int max_payload_size;                   // ++SCOTT
  30         rmr_mbuf_t*             sbuf;                                   // send buffer
  31         rmr_mbuf_t*             rbuf;                                   // received buffer
  32 };
  33
  34
  35 struct rmr_context * rmr_init_wrapper(char* lport){
  36
  37         struct rmr_context *rmr_c = malloc(sizeof (struct rmr_context));
  38         fprintf( stderr, "[INFO] glistening for replies on %s\n", lport );
  39
  40
  41         rmr_c->mrc = rmr_init(lport, RMR_MAX_RCV_BYTES, RMRFL_NONE );                   // setup RMr and get a context (rmr_c.mrc)
  42         rmr_c->ep_fd=-1;
  43
  44         //polling related initializations
  45         rmr_c->rcv_fd = rmr_get_rcvfd( rmr_c->mrc );// get the fd to poll for messages received
  46         if( rmr_c->rcv_fd < 0 ) {
  47                 fprintf( stderr, "[FAIL] unable to set up polling fd\n" );
  48                 exit( 1 );
  49         }
  50
  51         if( (rmr_c->ep_fd = epoll_create1( 0 )) < 0 ) {
  52                 fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
  53                 exit( 1 );
  54         }
  55         rmr_c->epe.events = EPOLLIN;
  56         rmr_c->epe.data.fd = rmr_c->rcv_fd;
  57
  58         if( epoll_ctl( rmr_c->ep_fd, EPOLL_CTL_ADD, rmr_c->rcv_fd, &rmr_c->epe ) != 0 )  {
  59                 fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
  60                 exit( 1 );
  61         }
  62         //end of polling related initializations
  63
  64         //++SCOTT------ next lines until !!SCOTT
  65         rmr_c->max_payload_size = 240;                                                                          //default
  66         if( (eparm = getenv( "DEMO_MAX_PAYLOAD_BYTES" )) != NULL ) {
  67                 rmr_c->max_payload_size = atoi(eparm);                                          // override with env
  68         }
  69         //!!SCOTT
  70
  71         //~~SCOTT next line
  72         rmr_c->sbuf = rmr_alloc_msg( rmr_c->mrc, rmr_c->max_payload_size );             // allocate largest payload to send
  73         rmr_c->rbuf = NULL;
  74
  75         return rmr_c;
  76 }
  77
  78 void rmr_send_wrapper (struct rmr_context *rmr_c, int mtype, char* message) {
  79         //--SCOTT int max_payload_size = 240; //default
  80         //--SCOTT if( (eparm = getenv( "DEMO_MAX_PAYLOAD_BYTES" )) != NULL ) {
  81                 //--SCOTT max_payload_size = atoi(eparm);
  82         //--SCOTT }
  83
  84         //~~SCOTT  next line
  85         snprintf( rmr_c->sbuf->payload, rmr_c->max_payload_size, "%s", message);                        // simple send message -- replace with real content
  86
  87         rmr_c->sbuf->mtype = mtype;                                                                             // fill in message meta data
  88         rmr_c->sbuf->len =  strlen( rmr_c->sbuf->payload ) + 1;                                 // actual length of payload (count the nil end of string)
  89         rmr_c->sbuf->state = 0;
  90
  91         //retry send for a few times before giving up
  92         long    natter = 0;                                             // natter on for errors only once in a while
  93         if( (rmr_c->sbuf = rmr_send_msg( rmr_c->mrc, rmr_c->sbuf )) != NULL ) {// unlikely, but might get a null pointer back if NNG really is buggered
  94                 if( rmr_c->sbuf->state != RMR_OK ) {
  95                         if( errno == EAGAIN ) {
  96                                 while( rmr_c->sbuf->state != RMR_OK && errno == EAGAIN ) {// NNG likes to refuse sends, just keep trying on eagain
  97                                         rmr_send_msg( rmr_c->mrc, rmr_c->sbuf );
  98                                 }
  99                         } else {// most likely connection refused, don't natter on
 100                                 if( time( NULL ) > natter ) {
 101                                         fprintf( stderr, "[WARN] send failed, pausing (%s)\n", strerror( errno ) );
 102                                         natter = time( NULL ) + 2;
 103                                 }
 104                                 sleep( 1 );
 105                         }
 106                 }
 107         } else {
 108                 if( time( NULL ) > natter ) {
 109                         fprintf( stderr, "[WARN] send failed, pausing (%s)\n", strerror( errno ) );
 110                         natter = time( NULL ) + 2;
 111                 }
 112                 sleep( 1 );
 113         }
 114
 115 }
 116
 117 int rmr_poll_for_message(struct rmr_context *rmr_c){
 118         int     nready;
 119         int     i;
 120         int return_type =0;
 121         //start polling the channel to read the acknowledgment
 122         //~~SCOTT  next line
 123
 124         nready = epoll_wait( rmr_c->ep_fd, rmr_c->events, 10, epoll_to );       // wait up to epoll_to ms for a response
 125         for( i = 0; i < nready && i < 10; i++ ) {           // loop through to find what is ready
 126                 if( rmr_c->events[i].data.fd == rmr_c->rcv_fd ) {             // RMr has something
 127                         errno = 0;
 128                         rmr_c->rbuf = rmr_rcv_msg( rmr_c->mrc, rmr_c->rbuf );                   // something ready; this should not block
 129                         if( rmr_c->rbuf ) {
 130         //                      fprintf( stderr, "<TEST> acknowledgment received:%s\n",rmr_c->rbuf->payload);
 131                                 return_type = 1; //the message has been acknowledged
 132                         }
 133                 }
 134         }
 135         return return_type;
 136 }
 137
 138 void rmr_close_wrapper (struct rmr_context *rmr_c){
 139
 140         rmr_close( rmr_c->mrc );
 141         free(rmr_c);
 142 }