PR feedback

Signed-off-by: Jean-Noël Moyne <[email protected]>

Author: jnmoyne

server/consumer.go

@@ -2274,14 +2274,14 @@ func configsEqualSansDelivery(a, b ConsumerConfig) bool {
 func (o *consumer) sendAckReply(subj string, err *ApiError) {
 	o.mu.RLock()
 	defer o.mu.RUnlock()
-	o.sendAckReplyLockHeld(subj, err)
+	o.sendAckReplyLocked(subj, err)
 }
 
 // Helper to send a reply to an ack.
 // Lock must be held
-func (o *consumer) sendAckReplyLockHeld(subj string, err *ApiError) {
+func (o *consumer) sendAckReplyLocked(subj string, err *ApiError) {
 
-	var resp = JSApiConsumerDeleteResponse{ApiResponse: ApiResponse{Type: JSApiConsumerAckResponseType, Error: err}}
+	var resp = JSApiConsumerAckResponse{ApiResponse: ApiResponse{Type: JSApiConsumerAckResponseType, Error: err}}
 
 	if err == nil {
 		resp.Success = true
@@ -2350,11 +2350,9 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) {
 
 	switch {
 	case len(msg) == 0, bytes.Equal(msg, AckAck), bytes.Equal(msg, AckOK):
-		if b, e := o.processAckMsg(sseq, dseq, dc, reply, true); e == nil {
-			if !b {
-				// We handle replies for acks in updateAcks
-				skipAckReply = true
-			}
+		if b, e := o.processAckMsg(sseq, dseq, dc, reply, true); e == nil && !b {
+			// We handle replies for acks in updateAcks
+			skipAckReply = true
 		} else {
 			err = e
 		}
@@ -2531,7 +2529,7 @@ func (o *consumer) updateAcks(dseq, sseq uint64, reply string, err *ApiError) {
 		o.store.UpdateAcks(dseq, sseq)
 		if reply != _EMPTY_ {
 			// Already locked so send direct.
-			o.sendAckReplyLockHeld(reply, err)
+			o.sendAckReplyLocked(reply, err)
 		}
 	}
 	// Update activity.

server/jetstream_cluster.go

@@ -5275,7 +5275,7 @@ func (o *consumer) processReplicatedAck(dseq, sseq uint64) error {
 
 	// Check if we have a reply that was requested.
 	if reply := o.replies[sseq]; reply != _EMPTY_ {
-		o.sendAckReplyLockHeld(reply, nil)
+		o.sendAckReplyLocked(reply, nil)
 		delete(o.replies, sseq)
 	}
 

server/jetstream_cluster_1_test.go

@@ -6714,14 +6714,15 @@ func TestJetStreamConsumerClusterAckAck(t *testing.T) {
 	require_NoError(t, err)
 	require_NoError(t, json.Unmarshal(replyMsg.Data, &resp))
 	require_False(t, resp.Success)
-	require_True(t, resp.Error.ErrCode == uint16(JSConsumerMsgNotPendingAckErr))
+	require_Equal(t, resp.Error.ErrCode, uint16(JSConsumerMsgNotPendingAckErr))
 
 	// Now ack way past the last sequence.
 	resp = JSApiConsumerAckResponse{}
 	replyMsg, err = nc.Request("$JS.ACK.TEST.C.1.10000000000.0.0.0", nil, 250*time.Millisecond)
 	require_NoError(t, err)
 	require_NoError(t, json.Unmarshal(replyMsg.Data, &resp))
-	require_True(t, resp.Error != nil && resp.Error.ErrCode == uint16(JSConsumerMsgNotPendingAckErr))
+	require_False(t, resp.Success)
+	require_Equal(t, resp.Error.ErrCode, uint16(JSConsumerMsgNotPendingAckErr))
 
 	// Make sure that no changes happened to our state.
 	ci, err := js.ConsumerInfo("TEST", "C")

server/jetstream_test.go

@@ -3033,14 +3033,16 @@ func TestJetStreamConsumerAckAck(t *testing.T) {
 	require_NoError(t, err)
 	resp = JSApiConsumerAckResponse{}
 	require_NoError(t, json.Unmarshal(ackReply2.Data, &resp))
-	require_True(t, !resp.Success && resp.Error.ErrCode == uint16(JSConsumerMsgNotPendingAckErr))
+	require_False(t, resp.Success)
+	require_Equal(t, resp.Error.ErrCode, uint16(JSConsumerMsgNotPendingAckErr))
 
 	// Error trying to ack an out of range sequence number
 	ackReply3, err := nc.Request("$JS.ACK.ACK-ACK.worker.1.6.0.0.0", AckAck, 10*time.Millisecond)
 	require_NoError(t, err)
 	resp = JSApiConsumerAckResponse{}
 	require_NoError(t, json.Unmarshal(ackReply3.Data, &resp))
-	require_True(t, !resp.Success && resp.Error.ErrCode == uint16(JSConsumerMsgNotPendingAckErr))
+	require_False(t, resp.Success)
+	require_Equal(t, resp.Error.ErrCode, uint16(JSConsumerMsgNotPendingAckErr))
 }
 
 func TestJetStreamAckNext(t *testing.T) {