mirror of
https://github.com/gogrlx/nats-server.git
synced 2026-04-02 03:38:42 -07:00
[FIXED] Possible panic in consumer, needed to recheck if consumer was closed (#4541)
Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
@@ -3794,7 +3794,9 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
|
||||
sz int
|
||||
wrn, wrb int
|
||||
)
|
||||
|
||||
o.mu.Lock()
|
||||
|
||||
// consumer is closed when mset is set to nil.
|
||||
if o.mset == nil {
|
||||
o.mu.Unlock()
|
||||
@@ -3817,6 +3819,13 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
|
||||
// Grab our next msg.
|
||||
pmsg, dc, err = o.getNextMsg()
|
||||
|
||||
// We can release the lock now under getNextMsg so need to check this condition again here.
|
||||
// consumer is closed when mset is set to nil.
|
||||
if o.mset == nil {
|
||||
o.mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
// On error either wait or return.
|
||||
if err != nil || pmsg == nil {
|
||||
// On EOF we can optionally fast sync num pending state.
|
||||
|
||||
@@ -5104,7 +5104,7 @@ func TestFileStoreNumPendingLargeNumBlks(t *testing.T) {
|
||||
|
||||
start = time.Now()
|
||||
total, _ = fs.NumPending(6000, "zzz", false)
|
||||
require_LessThan(t, time.Since(start), 15*time.Millisecond)
|
||||
require_LessThan(t, time.Since(start), 25*time.Millisecond)
|
||||
require_Equal(t, total, 4001)
|
||||
|
||||
// Now delete a message in first half and second half.
|
||||
|
||||
Reference in New Issue
Block a user