X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Frmr%2Fnng%2Fsrc%2Fmt_call_nng_static.c;h=78205611704bb8432f20189641b4629d85a2f42e;hb=7d1b7b0de82414ad25ebd323b2913de72ab448c1;hp=2ef21f67066bde68c8fc1308817edb721fc62f78;hpb=412d53dfa2f9b5b56a448797d0dfec3b0f11f666;p=ric-plt%2Flib%2Frmr.git

diff --git a/src/rmr/nng/src/mt_call_nng_static.c b/src/rmr/nng/src/mt_call_nng_static.c
index 2ef21f6..7820561 100644
--- a/src/rmr/nng/src/mt_call_nng_static.c
+++ b/src/rmr/nng/src/mt_call_nng_static.c
@@ -32,16 +32,21 @@
 #include <semaphore.h>
 
 static inline void queue_normal( uta_ctx_t* ctx, rmr_mbuf_t* mbuf ) {
+	static	int warned = 0;
 	chute_t*	chute;
-	int			state;
 
 	if( ! uta_ring_insert( ctx->mring, mbuf ) ) {
 		rmr_free_msg( mbuf );								// drop if ring is full
+		if( !warned ) {
+			fprintf( stderr, "[WARN] rmr_mt_receive: application is not receiving fast enough; messages dropping\n" );
+			warned++;
+		}
+
+		return;
 	}
 
 	chute = &ctx->chutes[0];
-	chute->mbuf = mbuf;
-	state = sem_post( &chute->barrier );								// tickle the ring monitor
+	sem_post( &chute->barrier );								// tickle the ring monitor
 }
 
 /*
@@ -59,7 +64,8 @@ static inline void queue_normal( uta_ctx_t* ctx, rmr_mbuf_t* mbuf ) {
 		The transaction ID in the message matches the expected ID in the chute,
 		then the message is given to the chute and the chute's semaphore is tickled.
 
-		If none are true, the message is dropped.
+		If none are true, the message is queued on the normal receive queue and that
+		related semaphore is tickeled.
 */
 static void* mt_receive( void* vctx ) {
 	uta_ctx_t*		ctx;
@@ -79,7 +85,7 @@ static void* mt_receive( void* vctx ) {
 	while( ! ctx->shutdown ) {
 		mbuf = rcv_msg( ctx, NULL );
 
-		if( mbuf != NULL && (hdr = (uta_mhdr_t *) mbuf->header) != NULL ) {
+		if( mbuf != NULL && (hdr = (uta_mhdr_t *) mbuf->header) != NULL && mbuf->payload != NULL ) {
 			if( hdr->flags & HFL_CALL_MSG ) {					// call generated message; ignore call-id etc and queue
 				queue_normal( ctx, mbuf );
 			} else {
@@ -91,13 +97,8 @@ static void* mt_receive( void* vctx ) {
 						queue_normal( ctx, mbuf );
 					} else {
 						chute = &ctx->chutes[call_id];
-						if( memcmp( mbuf->xaction, chute->expect, RMR_MAX_XID ) == 0 ) {		// match
-							chute->mbuf = mbuf;
-							sem_post( &chute->barrier );
-						} else {
-							rmr_free_msg( mbuf );
-							mbuf = NULL;
-						}
+						chute->mbuf = mbuf;
+						sem_post( &chute->barrier );				// the call function can vet xaction id in their own thread
 					}
 				}
 			}