- if( DEBUG ) fprintf( stderr, "[DBUG] data callback enough in the buffer size=%d need=%d remain=%d\n", river->msg_size, need, remain );
- memcpy( &river->accum[river->ipt], buf+bidx, need ); // grab just what is needed (might be more)
- buf2mbuf( ctx, river->accum, river->msg_size, fd ); // build an RMR mbuf and queue
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "data callback enough in the buffer size=%d need=%d remain=%d\n", river->msg_size, need, remain );
+ if( (river->flags & RF_DROP) == 0 ) {
+ memcpy( &river->accum[river->ipt], buf+bidx, need ); // grab just what is needed (might be more)
+ buf2mbuf( ctx, river->accum, river->msg_size, fd ); // build an RMR mbuf and queue
+ river->accum = (char *) malloc( sizeof( char ) * river->nbytes ); // fresh accumulator
+ } else {
+ if( !(river->flags & RF_NOTIFIED) ) {
+ rmr_vlog( RMR_VL_WARN, "message larger than max (%d) have arrived on fd %d\n", river->nbytes, fd );
+ river->flags |= RF_NOTIFIED;
+ }
+ }