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 very simple receiver that does nothing but listen
24 for messages and write stats every so often to the tty.
26 The receiver expects messages which have some trace information
27 and a message format of:
28 ck1 ck2|<msg text><nil>
30 ck1 is a simple checksum of the message text (NOT including the
31 nil at the end of the string.
33 ck2 is a simple checksum of the trace data which for the purposes
34 of testing is assumed to have a terminating nil to keep this simple.
36 Good messages are messages where both computed checksums match
37 the ck1 and ck2 values.
39 The receiver will send an 'ack' message back to the sender for
40 all type 5 messages received.
42 The sender and receiver can be run on the same host/container
43 or on different hosts. The route table is the key to setting
44 things up properly. See the sender code for rt information.
46 Define these environment variables to have some control:
47 RMR_SEED_RT -- path to the static routing table
48 RMR_RTG_SVC -- port to listen for RTG connections
51 Author: E. Scott Daniels
63 static int sum( char* str ) {
68 sum += *(str++) + i++;
75 Split the message at the first sep and return a pointer to the first
78 static char* split( char* str, char sep ) {
81 s = strchr( str, sep );
87 fprintf( stderr, "<RCVR> no pipe in message: (%s)\n", str );
91 int main( int argc, char** argv ) {
92 void* mrc; // msg router context
93 rmr_mbuf_t* msg = NULL; // message received
97 char* listen_port = "4560";
98 long count = 0; // total received
99 long good = 0; // good palyload buffers
100 long bad = 0; // payload buffers which were not correct
101 long bad_tr = 0; // trace buffers that were not correct
104 char wbuf[1024]; // we'll pull trace data into here
105 int nmsgs = 10; // number of messages to stop after (argv[1] overrides)
106 int rt_count = 0; // retry count
107 long ack_count = 0; // number of acks sent
109 data = getenv( "RMR_RTG_SVC" );
111 setenv( "RMR_RTG_SVC", "19289", 1 ); // set one that won't collide with the sender if on same host
115 nmsgs = atoi( argv[1] );
118 listen_port = argv[2];
121 fprintf( stderr, "<RCVR> listening on port: %s for a max of %d messages\n", listen_port, nmsgs );
123 mrc = rmr_init( listen_port, RMR_MAX_RCV_BYTES, RMRFL_NONE ); // start your engines!
125 fprintf( stderr, "<RCVR> ABORT: unable to initialise RMr\n" );
129 timeout = time( NULL ) + 20;
130 while( ! rmr_ready( mrc ) ) { // wait for RMr to load a route table
131 fprintf( stderr, "<RCVR> waiting for RMr to show ready\n" );
134 if( time( NULL ) > timeout ) {
135 fprintf( stderr, "<RCVR> giving up\n" );
139 fprintf( stderr, "<RCVR> rmr now shows ready, listening begins\n" );
141 timeout = time( NULL ) + 20;
142 while( count < nmsgs ) {
143 msg = rmr_torcv_msg( mrc, msg, 1000 ); // wait for about 1s so that if sender never starts we eventually escape
146 if( msg->state == RMR_OK ) {
147 if( (data = split( msg->payload, '|' )) != NULL ) {
148 if( sum( data ) == atoi( (char *) msg->payload ) ) {
151 fprintf( stderr, "<RCVR> chk sum bad: computed=%d expected;%d (%s)\n", sum( data ), atoi( msg->payload ), data );
156 if( (data = split( msg->payload, ' ' )) != NULL ) { // data will point to the chksum for the trace data
157 state = rmr_get_trace( msg, wbuf, 1024 ); // should only copy upto the trace size; we'll check that
158 if( state > 128 || state < 1 ) {
159 fprintf( stderr, "trace data size listed unexpectedly long: %d\n", state );
161 if( sum( wbuf ) != atoi( data ) ) {
162 fprintf( stderr, "<RCVR> trace chk sum bad: computed=%d expected;%d len=%d (%s)\n", sum( wbuf ), atoi( data ), state, wbuf );
166 count++; // messages received for stats output
168 if( msg->mtype == 5 ) { // send an ack; sender will count but not process, so data in message is moot
169 msg = rmr_rts_msg( mrc, msg ); // we don't try to resend if this returns retry
171 while( rt_count > 0 && msg != NULL && msg->state == RMR_ERR_RETRY ) { // to work right in nano we need this :(
172 if( ack_count < 1 ) { // need to connect, so hard wait
176 msg = rmr_rts_msg( mrc, msg ); // we don't try to resend if this returns retry
178 if( msg && msg->state == RMR_OK ) { // if it eventually worked
185 if( time( NULL ) > timeout ) {
186 fprintf( stderr, "receiver timed out\n" );
192 fprintf( stderr, "<RCVR> [%s] %ld messages; good=%ld acked=%ld bad=%ld bad-trace=%ld\n", !!(errors + bad + bad_tr) ? "FAIL" : "PASS", count, good, ack_count, bad, bad_tr );
193 sleep( 2 ); // let any outbound acks flow before closing
196 return !!(errors + bad + bad_tr); // bad rc if any are !0