1 // :vim ts=4 sw=4 noet:
3 ==================================================================================
4 Copyright (c) 2019 Nokia
5 Copyright (c) 2018-2019 AT&T Intellectual Property.
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
11 http://www.apache.org/licenses/LICENSE-2.0
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 ==================================================================================
23 Abstract: This is a simple sender which will send a series of messages.
24 It is expected that the first attempt(s) will fail if the receiver
25 is not up and this does not start decrementing the number to
26 send until it has a good send.
28 The process will check the receive queue and list received messages
29 but pass/fail is not dependent on what comes back.
31 If the receiver(s) do not become connectable in 20 sec this process
32 will give up and fail.
35 Message types will vary between 0 and 9, so the route table must
36 be set up to support those message types. Further, for message types
37 0, 1 and 2, the subscription ID will be set to type x 10, so the route
38 table must be set to include the sub-id for those types in order for
39 the messages to reach their destination.
42 ck1 ck2|<msg-txt><nil>
44 Ck1 is the simple check sum of the msg-text (NOT includeing <nil>)
45 Ck2 is the simple check sum of the trace data which is a nil terminated
48 Parms: argv[1] == number of msgs to send (10)
49 argv[2] == delay (mu-seconds, 1000000 default)
50 argv[3] == max msg type (not inclusive; default 10)
51 argv[4] == listen port
53 Sender will send for at most 20 seconds, so if nmsgs and delay extend
54 beyond that period the total number of messages sent will be less
58 Author: E. Scott Daniels
66 #include <sys/epoll.h>
71 static int sum( char* str ) {
76 sum += *(str++) + i++;
82 int main( int argc, char** argv ) {
83 void* mrc; // msg router context
84 struct epoll_event events[1]; // list of events to give to epoll
85 struct epoll_event epe; // event definition for event to listen to
86 int ep_fd = -1; // epoll's file des (given to epoll_wait)
87 int rcv_fd; // file des that NNG tickles -- give this to epoll to listen on
88 int nready; // number of events ready for receive
89 rmr_mbuf_t* sbuf; // send buffer
90 rmr_mbuf_t* rbuf; // received buffer
92 int rt_count = 0; // number of messages requiring a spin retry
94 int fail_count = 0; // # of failure sends after first successful send
95 char* listen_port = "43086";
98 int successful = 0; // set to true after we have a successful send
102 int delay = 100000; // usec between send attempts
103 int nmsgs = 10; // number of messages to send
104 int max_mt = 10; // reset point for message type
107 nmsgs = atoi( argv[1] );
110 delay = atoi( argv[2] );
113 max_mt = atoi( argv[3] );
116 listen_port = argv[4];
119 fprintf( stderr, "<SNDR> listen port: %s; sending %d messages; delay=%d\n", listen_port, nmsgs, delay );
121 if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
122 fprintf( stderr, "<SNDR> unable to initialise RMr\n" );
126 if( (rcv_fd = rmr_get_rcvfd( mrc )) >= 0 ) { // epoll only available from NNG -- skip receive later if not NNG
128 fprintf( stderr, "<SNDR> unable to set up polling fd\n" );
131 if( (ep_fd = epoll_create1( 0 )) < 0 ) {
132 fprintf( stderr, "<SNDR> [FAIL] unable to create epoll fd: %d\n", errno );
135 epe.events = EPOLLIN;
136 epe.data.fd = rcv_fd;
138 if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 ) {
139 fprintf( stderr, "<SNDR> [FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
143 rmr_set_rtimeout( mrc, 0 ); // for nano we must set the receive timeout to 0; non-blocking receive
146 sbuf = rmr_alloc_msg( mrc, 512 ); // alloc first send buffer; subsequent buffers allcoated on send
147 //sbuf = rmr_tralloc_msg( mrc, 512, 11, "xxxxxxxxxx" ); // alloc first send buffer; subsequent buffers allcoated on send
148 rbuf = NULL; // don't need to alloc receive buffer
150 timeout = time( NULL ) + 20; // give rmr 20s to find the route table (shouldn't need that much)
151 while( ! rmr_ready( mrc ) ) { // must have a route table before we can send; wait til RMr says it has one
152 fprintf( stderr, "<SNDR> waiting for rmr to show ready\n" );
155 if( time( NULL ) > timeout ) {
156 fprintf( stderr, "<SNDR> giving up\n" );
160 fprintf( stderr, "<SNDR> rmr is ready; starting to send\n" );
162 timeout = time( NULL ) + 20;
164 while( count < nmsgs ) { // we send n messages after the first message is successful
165 snprintf( trace, 100, "%lld", (long long) time( NULL ) );
166 rmr_set_trace( sbuf, trace, strlen( trace ) + 1 );
167 snprintf( wbuf, 200, "count=%d tr=%s %d stand up and cheer!", count, trace, rand() );
168 snprintf( sbuf->payload, 300, "%d %d|%s", sum( wbuf ), sum( trace ), wbuf );
170 sbuf->mtype = mtype; // fill in the message bits
172 sbuf->sub_id = mtype * 10;
177 sbuf->len = strlen( sbuf->payload ) + 1; // our receiver likely wants a nice acsii-z string
179 sbuf = rmr_send_msg( mrc, sbuf ); // send it (send returns an empty payload on success, or the original payload on fail/retry)
181 switch( sbuf->state ) {
184 while( sbuf->state == RMR_ERR_RETRY ) { // soft failure (device busy?) retry
185 sbuf = rmr_send_msg( mrc, sbuf ); // retry send until it's good (simple test; real programmes should do better)
187 if( sbuf->state == RMR_OK ) {
188 successful = 1; // indicates only that we sent one successful message, not the current state
191 fail_count++; // count failures after first successful message
202 fail_count++; // count failures after first successful message
204 // some error (not connected likely), don't count this
209 if( successful ) { // once we have a message that was sent, start to increase things
212 if( mtype >= max_mt ) { // if large number of sends don't require infinite rt entries :)
218 while( (nready = epoll_wait( ep_fd, events, 1, 0 )) > 0 ) { // if something ready to receive (non-blocking check)
219 if( events[0].data.fd == rcv_fd ) { // we only are waiting on 1 thing, so [0] is ok
221 rbuf = rmr_rcv_msg( mrc, rbuf );
227 } else { // nano, we will only pick up one at a time.
228 if( (rbuf = rmr_rcv_msg( mrc, rbuf ) ) != NULL ) {
229 if( rbuf->state == RMR_OK ) {
235 if( time( NULL ) > timeout ) { // should only happen if we never connect or nmsg > what we can send in 20sec
236 fprintf( stderr, "sender timeout\n" );
246 timeout = time( NULL ) + 2; // allow 2 seconds for the pipe to drain from the receiver
247 while( time( NULL ) < timeout );
249 while( (nready = epoll_wait( ep_fd, events, 1, 100 )) > 0 ) { // if something ready to receive (non-blocking check)
250 if( events[0].data.fd == rcv_fd ) { // we only are waiting on 1 thing, so [0] is ok
252 rbuf = rmr_rcv_msg( mrc, rbuf );
255 timeout = time( NULL ) + 2;
259 } else { // nano, we will only pick up one at a time.
260 if( (rbuf = rmr_torcv_msg( mrc, rbuf, 100 ) ) != NULL ) {
261 if( rbuf->state == RMR_OK ) {
267 fprintf( stderr, "<SNDR> [%s] sent=%d rcvd-acks=%d failures=%d retries=%d\n", count == nmsgs ? "PASS" : "FAIL", count, rcvd_count, fail_count, rt_count );
270 return !( count == nmsgs );