avoid changing Receive behavior

client.go

@@ -1953,6 +1953,9 @@ type Receiver struct {
 
 // HandleMessage takes in a func to handle the incoming message.
 // Blocks until a message is received, ctx completes, or an error occurs.
+
+// Note: You must take an action on the message in the provided handler (Accept/Reject/Release/Modify)
+// or the pending message tracker will get out of sync, and reduce the flow.
 func (r *Receiver) HandleMessage(ctx context.Context, handle func(*Message) error) error {
 	debug(3, "Entering link %s Receive()", r.link.key.name)
 
@@ -2002,6 +2005,7 @@ func (r *Receiver) HandleMessage(ctx context.Context, handle func(*Message) erro
 	}
 }
 
+// Deprecated: prefer HandleMessage
 // Receive returns the next message from the sender.
 //
 // Blocks until a message is received, ctx completes, or an error occurs.
@@ -2013,24 +2017,14 @@ func (r *Receiver) Receive(ctx context.Context) (*Message, error) {
 		}
 	}
 
-	trackCompletion := func(msg *Message) {
-		<-msg.doneSignal
-		r.link.deletePending(msg)
-		debug(3, "Receive() deleted pending %d", msg.deliveryID)
-		if atomic.LoadUint32(&r.link.paused) == 1 {
-			select {
-			case r.link.receiverReady <- struct{}{}:
-				debug(3, "Receive() unpause link on completion")
-			default:
-			}
-		}
-	}
-
 	// non-blocking receive to ensure buffered messages are
 	// delivered regardless of whether the link has been closed.
 	select {
 	case msg := <-r.link.messages:
-		go trackCompletion(&msg)
+		// we remove the message from pending as soon as it's popped off the channel
+		// This makes the pending count the same as messages buffer count
+		// and keeps the behavior the same as before the pending messages tracking was introduced
+		defer r.link.deletePending(&msg)
 		debug(3, "Receive() non blocking %d", msg.deliveryID)
 		msg.receiver = r
 		return &msg, nil
@@ -2042,7 +2036,10 @@ func (r *Receiver) Receive(ctx context.Context) (*Message, error) {
 	// wait for the next message
 	select {
 	case msg := <-r.link.messages:
-		go trackCompletion(&msg)
+		// we remove the message from pending as soon as it's popped off the channel
+		// This makes the pending count the same as messages buffer count
+		// and keeps the behavior the same as before the pending messages tracking was introduced
+		defer r.link.deletePending(&msg)
 		debug(3, "Receive() blocking %d", msg.deliveryID)
 		msg.receiver = r
 		return &msg, nil