test/app_test/sender.c - oransc/ric-plt/lib/rmr - Gitiles

 // :vim ts=4 sw=4 noet:
 /*
 ==================================================================================
 	Copyright (c) 2019 Nokia
 	Copyright (c) 2018-2019 AT&T Intellectual Property.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

 	   http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 ==================================================================================
 */

 /*
 	Mnemonic:	sender.c
 	Abstract:	This is a simple sender which will send a series of messages.
 				It is expected that the first attempt(s) will fail if the receiver
 				is not up and this does not start decrementing the number to
 				send until it has a good send.

 				The process will check the receive queue and list received messages
 				but pass/fail is not dependent on what comes back.

 				If the receiver(s) do not become connectable in 20 sec this process
 				will give up and fail.


 				Message types will vary between 0 and 9, so the route table must
 				be set up to support those message types. Further, for message types
 				0, 1 and 2, the subscription ID will be set to type x 10, so the route
 				table must be set to include the sub-id for those types in order for
 				the messages to reach their destination.

 				Message format is:
 					ck1 ck2|<msg-txt><nil>

 				Ck1 is the simple check sum of the msg-text (NOT includeing <nil>)
 				Ck2 is the simple check sum of the trace data which is a nil terminated
 				series of bytes.

 				Parms:	argv[1] == number of msgs to send (10)
 						argv[2] == delay		(mu-seconds, 1000000 default)
 						argv[3] == max msg type (not inclusive; default 10)
 						argv[4] == listen port

 				Sender will send for at most 20 seconds, so if nmsgs and delay extend
 				beyond that period the total number of messages sent will be less
 				than n.

 	Date:		18 April 2019
 	Author:		E. Scott Daniels
 */

 #include <unistd.h>
 #include <errno.h>
 #include <string.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/epoll.h>
 #include <time.h>

 #include <rmr/rmr.h>

 #define WBUF_SIZE	1024
 #define TRACE_SIZE	1024

 static int sum( char* str ) {
 	int sum = 0;
 	int	i = 0;

 	while( *str ) {
 		sum += *(str++) + i++;
 	}

 	return sum % 255;
 }

 /*
 	See if my id string is in the buffer immediately after the first >.
 	Return 1 if so, 0 if not.
 */
 static int vet_received( char* me, char* buf ) {
 	char*	ch;

 	if( (ch = strchr( buf, '>' )) == NULL ) {
 		return 0;
 	}

 	return strcmp( me, ch+1 ) == 0;
 }

 int main( int argc, char** argv ) {
 	void* mrc;      						// msg router context
 	struct	epoll_event events[1];			// list of events to give to epoll
 	struct	epoll_event epe;				// event definition for event to listen to
 	int     ep_fd = -1;						// epoll's file des (given to epoll_wait)
 	int		rcv_fd;    						// file des that NNG tickles -- give this to epoll to listen on
 	int		nready;							// number of events ready for receive
 	rmr_mbuf_t*		sbuf;					// send buffer
 	rmr_mbuf_t*		rbuf;					// received buffer
 	char*	ch;
 	int		count = 0;
 	int		rt_count = 0;					// number of messages requiring a spin retry
 	int		rcvd_count = 0;
 	int		rts_ok = 0;						// number received with our tag
 	int		fail_count = 0;					// # of failure sends after first successful send
 	char*	listen_port = "43086";
 	int		mtype = 0;
 	int		stats_freq = 100;
 	int		successful = 0;					// set to true after we have a successful send
 	char*	wbuf = NULL;					// working buffer
 	char	me[128];						// who I am to vet rts was actually from me
 	char*	trace = NULL;					// area to build trace data in
 	long	timeout = 0;
 	int		delay = 100000;					// usec between send attempts
 	int		nmsgs = 10;						// number of messages to send
 	int		max_mt = 10;					// reset point for message type
 	int		start_mt = 0;
 	int		pass = 1;

 	wbuf = (char *) malloc( sizeof( char ) * WBUF_SIZE );
 	trace = (char *) malloc( sizeof( char ) * TRACE_SIZE );

 	if( argc > 1 ) {
 		nmsgs = atoi( argv[1] );
 	}
 	if( argc > 2 ) {
 		delay = atoi( argv[2] );
 	}
 	if( argc > 3 ) {
 		if( (ch = strchr( argv[3], ':' )) != NULL ) {
 			max_mt = atoi( ch+1 );
 			start_mt = atoi( argv[3] );
 		} else {
 			max_mt = atoi( argv[3] );
 		}
 	}
 	if( argc > 4 ) {
 		listen_port = argv[4];
 	}

 	mtype = start_mt;

 	fprintf( stderr, "<SNDR> listen port: %s; sending %d messages; delay=%d\n", listen_port, nmsgs, delay );

 	if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
 		fprintf( stderr, "<SNDR> unable to initialise RMr\n" );
 		exit( 1 );
 	}

 	if( (rcv_fd = rmr_get_rcvfd( mrc )) >= 0 ) {			// epoll only available from NNG -- skip receive later if not NNG
 		if( rcv_fd < 0 ) {
 			fprintf( stderr, "<SNDR> unable to set up polling fd\n" );
 			exit( 1 );
 		}
 		if( (ep_fd = epoll_create1( 0 )) < 0 ) {
 			fprintf( stderr, "<SNDR> [FAIL] unable to create epoll fd: %d\n", errno );
 			exit( 1 );
 		}
 		epe.events = EPOLLIN;
 		epe.data.fd = rcv_fd;

 		if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 			fprintf( stderr, "<SNDR> [FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
 			exit( 1 );
 		}
 	} else {
 		rmr_set_rtimeout( mrc, 0 );			// for nano we must set the receive timeout to 0; non-blocking receive
 	}

 	sbuf = rmr_alloc_msg( mrc, 1024 );	// alloc first send buffer; subsequent buffers allcoated on send
 	//sbuf = rmr_tralloc_msg( mrc, 1024, 11, "xxxxxxxxxx" );	// alloc first send buffer; subsequent buffers allcoated on send
 	rbuf = NULL;						// don't need to alloc receive buffer

 	timeout = time( NULL ) + 20;		// give rmr 20s to find the route table (shouldn't need that much)
 	while( ! rmr_ready( mrc ) ) {		// must have a route table before we can send; wait til RMr says it has one
 		fprintf( stderr, "<SNDR> waiting for rmr to show ready\n" );
 		sleep( 1 );

 		if( time( NULL ) > timeout ) {
 			fprintf( stderr, "<SNDR> giving up\n" );
 			exit( 1 );
 		}
 	}
 	fprintf( stderr, "<SNDR> rmr is ready; starting to send\n" );

 	timeout = time( NULL ) + 20;

 	gethostname( wbuf, WBUF_SIZE );
 	snprintf( me, sizeof( me ), "%s-%d", wbuf, getpid( ) );

 	while( count < nmsgs ) {								// we send n messages after the first message is successful
 		snprintf( trace, 100, "%lld", (long long) time( NULL ) );
 		rmr_set_trace( sbuf, trace, strlen( trace ) + 1 );
 		snprintf( wbuf, 512, "count=%d tr=%s %d stand up and cheer!>%s", count, trace, rand(), me );
 		snprintf( sbuf->payload, 1024, "%d %d|%s", sum( wbuf ), sum( trace ), wbuf );

 		sbuf->mtype = mtype;							// fill in the message bits
 		if( mtype < 3 ) {
 			sbuf->sub_id = mtype * 10;
 		} else {
 			sbuf->sub_id = -1;
 		}

 		sbuf->len =  strlen( sbuf->payload ) + 1;		// our receiver likely wants a nice acsii-z string
 		sbuf->state = 0;

 		fprintf( stderr, "<SNDR> sending msg type %d\n", sbuf->mtype );

 		sbuf = rmr_send_msg( mrc, sbuf );				// send it (send returns an empty payload on success, or the original payload on fail/retry)

 		switch( sbuf->state ) {
 			case RMR_ERR_RETRY:
 				rt_count++;
 				while( time( NULL ) < timeout && sbuf->state == RMR_ERR_RETRY ) {			// soft failure (device busy?) retry
 					sbuf = rmr_send_msg( mrc, sbuf );			// retry send until it's good (simple test; real programmes should do better)
 				}
 				if( sbuf->state == RMR_OK ) {
 					if( successful == 0 ) {
 						fail_count = 0;							// count only after first message goes through
 					}
 					successful = 1; 							// indicates only that we sent one successful message, not the current state
 				} else {
 					fail_count++;							// count failures after first successful message
 					if( !successful && fail_count > 30 ) {
 						fprintf( stderr, "[FAIL] too many send errors for this test\n" );
 						exit( 1 );
 					}
 				}
 				break;

 			case RMR_OK:
 				successful = 1;
 				break;

 			default:
 				if( successful ) {
 					fail_count++;							// count failures after first successful message
 				}
 				// some error (not connected likely), don't count this
 				//sleep( 1 );
 				break;
 		}

 		if( successful ) {				// once we have a message that was sent, start to increase things
 			count++;
 			mtype++;
 			if( mtype >= max_mt ) {			// if large number of sends don't require infinite rt entries :)
 				mtype = start_mt;
 			}
 		}

 		if( rcv_fd >= 0 ) {
 			while( (nready = epoll_wait( ep_fd, events, 1, 0 )) > 0 ) {			// if something ready to receive (non-blocking check)
 				if( events[0].data.fd == rcv_fd ) {             				// we only are waiting on 1 thing, so [0] is ok
 					errno = 0;
 					rbuf = rmr_rcv_msg( mrc, rbuf );
 					if( rbuf ) {
 						rts_ok += vet_received( me, rbuf->payload );
 						rcvd_count++;
 					}
 				}
 			}
 		} else {				// nano, we will only pick up one at a time.
 			if(	(rbuf = rmr_rcv_msg( mrc, rbuf ) ) != NULL ) {
 				if( rbuf->state == RMR_OK ) {
 					rts_ok += vet_received( me, rbuf->payload );
 					rcvd_count++;
 				}
 			}
 		}

 		if( time( NULL ) > timeout ) {						// should only happen if we never connect or nmsg > what we can send in 20sec
 			fprintf( stderr, "sender timeout\n" );
 			break;
 		}

 		if( delay > 0 ) {
 			usleep( delay );
 		}
 	}

 	fprintf( stderr, "<SNDR> draining begins\n" );
 	timeout = time( NULL ) + 10;				// allow 10 seconds for the pipe to drain from the receiver
 	while( time( NULL ) < timeout ) {
 		if( rcv_fd >= 0 ) {
 			while( (nready = epoll_wait( ep_fd, events, 1, 100 )) > 0 ) {			// if something ready to receive (non-blocking check)
 				if( events[0].data.fd == rcv_fd ) {             				// we only are waiting on 1 thing, so [0] is ok
 					errno = 0;
 					rbuf = rmr_rcv_msg( mrc, rbuf );
 					if( rbuf ) {
 						rcvd_count++;
 						rts_ok += vet_received( me, rbuf->payload );
 						timeout = time( NULL ) + 10;							// break 10s after last received message
 					}
 				}
 			}
 		} else {				// nano, we will only pick up one at a time.
 			if(	(rbuf = rmr_torcv_msg( mrc, rbuf, 100 ) ) != NULL ) {
 				if( rbuf->state == RMR_OK ) {
 					rcvd_count++;
 					rts_ok += vet_received( me, rbuf->payload );
 				}
 			}
 		}
 	}
 	fprintf( stderr, "<SNDR> draining finishes\n" );

 	if( rcvd_count != rts_ok || count != nmsgs ) {
 		pass = 0;
 	}

 	fprintf( stderr, "<SNDR> [%s] sent=%d  rcvd=%d  rts-ok=%d failures=%d retries=%d\n",
 		pass ? "PASS" : "FAIL",  count, rcvd_count, rts_ok, fail_count, rt_count );
 	rmr_close( mrc );

 	return !( count == nmsgs );
 }
	// :vim ts=4 sw=4 noet:
	/*
	==================================================================================
	Copyright (c) 2019 Nokia
	Copyright (c) 2018-2019 AT&T Intellectual Property.

	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
	==================================================================================
	*/

	/*
	Mnemonic: sender.c
	Abstract: This is a simple sender which will send a series of messages.
	It is expected that the first attempt(s) will fail if the receiver
	is not up and this does not start decrementing the number to
	send until it has a good send.

	The process will check the receive queue and list received messages
	but pass/fail is not dependent on what comes back.

	If the receiver(s) do not become connectable in 20 sec this process
	will give up and fail.


	Message types will vary between 0 and 9, so the route table must
	be set up to support those message types. Further, for message types
	0, 1 and 2, the subscription ID will be set to type x 10, so the route
	table must be set to include the sub-id for those types in order for
	the messages to reach their destination.

	Message format is:
	ck1 ck2\|<msg-txt><nil>

	Ck1 is the simple check sum of the msg-text (NOT includeing <nil>)
	Ck2 is the simple check sum of the trace data which is a nil terminated
	series of bytes.

	Parms: argv[1] == number of msgs to send (10)
	argv[2] == delay (mu-seconds, 1000000 default)
	argv[3] == max msg type (not inclusive; default 10)
	argv[4] == listen port

	Sender will send for at most 20 seconds, so if nmsgs and delay extend
	beyond that period the total number of messages sent will be less
	than n.

	Date: 18 April 2019
	Author: E. Scott Daniels
	*/

	#include <unistd.h>
	#include <errno.h>
	#include <string.h>
	#include <stdio.h>
	#include <stdlib.h>
	#include <sys/epoll.h>
	#include <time.h>

	#include <rmr/rmr.h>

	#define WBUF_SIZE 1024
	#define TRACE_SIZE 1024

	static int sum( char* str ) {
	int sum = 0;
	int i = 0;

	while( *str ) {
	sum += *(str++) + i++;
	}

	return sum % 255;
	}

	/*
	See if my id string is in the buffer immediately after the first >.
	Return 1 if so, 0 if not.
	*/
	static int vet_received( char* me, char* buf ) {
	char* ch;

	if( (ch = strchr( buf, '>' )) == NULL ) {
	return 0;
	}

	return strcmp( me, ch+1 ) == 0;
	}

	int main( int argc, char** argv ) {
	void* mrc; // msg router context
	struct epoll_event events[1]; // list of events to give to epoll
	struct epoll_event epe; // event definition for event to listen to
	int ep_fd = -1; // epoll's file des (given to epoll_wait)
	int rcv_fd; // file des that NNG tickles -- give this to epoll to listen on
	int nready; // number of events ready for receive
	rmr_mbuf_t* sbuf; // send buffer
	rmr_mbuf_t* rbuf; // received buffer
	char* ch;
	int count = 0;
	int rt_count = 0; // number of messages requiring a spin retry
	int rcvd_count = 0;
	int rts_ok = 0; // number received with our tag
	int fail_count = 0; // # of failure sends after first successful send
	char* listen_port = "43086";
	int mtype = 0;
	int stats_freq = 100;
	int successful = 0; // set to true after we have a successful send
	char* wbuf = NULL; // working buffer
	char me[128]; // who I am to vet rts was actually from me
	char* trace = NULL; // area to build trace data in
	long timeout = 0;
	int delay = 100000; // usec between send attempts
	int nmsgs = 10; // number of messages to send
	int max_mt = 10; // reset point for message type
	int start_mt = 0;
	int pass = 1;

	wbuf = (char ) malloc( sizeof( char ) WBUF_SIZE );
	trace = (char ) malloc( sizeof( char ) TRACE_SIZE );

	if( argc > 1 ) {
	nmsgs = atoi( argv[1] );
	}
	if( argc > 2 ) {
	delay = atoi( argv[2] );
	}
	if( argc > 3 ) {
	if( (ch = strchr( argv[3], ':' )) != NULL ) {
	max_mt = atoi( ch+1 );
	start_mt = atoi( argv[3] );
	} else {
	max_mt = atoi( argv[3] );
	}
	}
	if( argc > 4 ) {
	listen_port = argv[4];
	}

	mtype = start_mt;

	fprintf( stderr, "<SNDR> listen port: %s; sending %d messages; delay=%d\n", listen_port, nmsgs, delay );

	if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
	fprintf( stderr, "<SNDR> unable to initialise RMr\n" );
	exit( 1 );
	}

	if( (rcv_fd = rmr_get_rcvfd( mrc )) >= 0 ) { // epoll only available from NNG -- skip receive later if not NNG
	if( rcv_fd < 0 ) {
	fprintf( stderr, "<SNDR> unable to set up polling fd\n" );
	exit( 1 );
	}
	if( (ep_fd = epoll_create1( 0 )) < 0 ) {
	fprintf( stderr, "<SNDR> [FAIL] unable to create epoll fd: %d\n", errno );
	exit( 1 );
	}
	epe.events = EPOLLIN;
	epe.data.fd = rcv_fd;

	if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 ) {
	fprintf( stderr, "<SNDR> [FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
	exit( 1 );
	}
	} else {
	rmr_set_rtimeout( mrc, 0 ); // for nano we must set the receive timeout to 0; non-blocking receive
	}

	sbuf = rmr_alloc_msg( mrc, 1024 ); // alloc first send buffer; subsequent buffers allcoated on send
	//sbuf = rmr_tralloc_msg( mrc, 1024, 11, "xxxxxxxxxx" ); // alloc first send buffer; subsequent buffers allcoated on send
	rbuf = NULL; // don't need to alloc receive buffer

	timeout = time( NULL ) + 20; // give rmr 20s to find the route table (shouldn't need that much)
	while( ! rmr_ready( mrc ) ) { // must have a route table before we can send; wait til RMr says it has one
	fprintf( stderr, "<SNDR> waiting for rmr to show ready\n" );
	sleep( 1 );

	if( time( NULL ) > timeout ) {
	fprintf( stderr, "<SNDR> giving up\n" );
	exit( 1 );
	}
	}
	fprintf( stderr, "<SNDR> rmr is ready; starting to send\n" );

	timeout = time( NULL ) + 20;

	gethostname( wbuf, WBUF_SIZE );
	snprintf( me, sizeof( me ), "%s-%d", wbuf, getpid( ) );

	while( count < nmsgs ) { // we send n messages after the first message is successful
	snprintf( trace, 100, "%lld", (long long) time( NULL ) );
	rmr_set_trace( sbuf, trace, strlen( trace ) + 1 );
	snprintf( wbuf, 512, "count=%d tr=%s %d stand up and cheer!>%s", count, trace, rand(), me );
	snprintf( sbuf->payload, 1024, "%d %d\|%s", sum( wbuf ), sum( trace ), wbuf );

	sbuf->mtype = mtype; // fill in the message bits
	if( mtype < 3 ) {
	sbuf->sub_id = mtype * 10;
	} else {
	sbuf->sub_id = -1;
	}

	sbuf->len = strlen( sbuf->payload ) + 1; // our receiver likely wants a nice acsii-z string
	sbuf->state = 0;

	fprintf( stderr, "<SNDR> sending msg type %d\n", sbuf->mtype );

	sbuf = rmr_send_msg( mrc, sbuf ); // send it (send returns an empty payload on success, or the original payload on fail/retry)

	switch( sbuf->state ) {
	case RMR_ERR_RETRY:
	rt_count++;
	while( time( NULL ) < timeout && sbuf->state == RMR_ERR_RETRY ) { // soft failure (device busy?) retry
	sbuf = rmr_send_msg( mrc, sbuf ); // retry send until it's good (simple test; real programmes should do better)
	}
	if( sbuf->state == RMR_OK ) {
	if( successful == 0 ) {
	fail_count = 0; // count only after first message goes through
	}
	successful = 1; // indicates only that we sent one successful message, not the current state
	} else {
	fail_count++; // count failures after first successful message
	if( !successful && fail_count > 30 ) {
	fprintf( stderr, "[FAIL] too many send errors for this test\n" );
	exit( 1 );
	}
	}
	break;

	case RMR_OK:
	successful = 1;
	break;

	default:
	if( successful ) {
	fail_count++; // count failures after first successful message
	}
	// some error (not connected likely), don't count this
	//sleep( 1 );
	break;
	}

	if( successful ) { // once we have a message that was sent, start to increase things
	count++;
	mtype++;
	if( mtype >= max_mt ) { // if large number of sends don't require infinite rt entries :)
	mtype = start_mt;
	}
	}

	if( rcv_fd >= 0 ) {
	while( (nready = epoll_wait( ep_fd, events, 1, 0 )) > 0 ) { // if something ready to receive (non-blocking check)
	if( events[0].data.fd == rcv_fd ) { // we only are waiting on 1 thing, so [0] is ok
	errno = 0;
	rbuf = rmr_rcv_msg( mrc, rbuf );
	if( rbuf ) {
	rts_ok += vet_received( me, rbuf->payload );
	rcvd_count++;
	}
	}
	}
	} else { // nano, we will only pick up one at a time.
	if( (rbuf = rmr_rcv_msg( mrc, rbuf ) ) != NULL ) {
	if( rbuf->state == RMR_OK ) {
	rts_ok += vet_received( me, rbuf->payload );
	rcvd_count++;
	}
	}
	}

	if( time( NULL ) > timeout ) { // should only happen if we never connect or nmsg > what we can send in 20sec
	fprintf( stderr, "sender timeout\n" );
	break;
	}

	if( delay > 0 ) {
	usleep( delay );
	}
	}

	fprintf( stderr, "<SNDR> draining begins\n" );
	timeout = time( NULL ) + 10; // allow 10 seconds for the pipe to drain from the receiver
	while( time( NULL ) < timeout ) {
	if( rcv_fd >= 0 ) {
	while( (nready = epoll_wait( ep_fd, events, 1, 100 )) > 0 ) { // if something ready to receive (non-blocking check)
	if( events[0].data.fd == rcv_fd ) { // we only are waiting on 1 thing, so [0] is ok
	errno = 0;
	rbuf = rmr_rcv_msg( mrc, rbuf );
	if( rbuf ) {
	rcvd_count++;
	rts_ok += vet_received( me, rbuf->payload );
	timeout = time( NULL ) + 10; // break 10s after last received message
	}
	}
	}
	} else { // nano, we will only pick up one at a time.
	if( (rbuf = rmr_torcv_msg( mrc, rbuf, 100 ) ) != NULL ) {
	if( rbuf->state == RMR_OK ) {
	rcvd_count++;
	rts_ok += vet_received( me, rbuf->payload );
	}
	}
	}
	}
	fprintf( stderr, "<SNDR> draining finishes\n" );

	if( rcvd_count != rts_ok \|\| count != nmsgs ) {
	pass = 0;
	}

	fprintf( stderr, "<SNDR> [%s] sent=%d rcvd=%d rts-ok=%d failures=%d retries=%d\n",
	pass ? "PASS" : "FAIL", count, rcvd_count, rts_ok, fail_count, rt_count );
	rmr_close( mrc );

	return !( count == nmsgs );
	}