@@ -30,8 +30,7 @@ func TestDelayQueue_consume(t *testing.T) {
3030 WithFetchInterval (time .Millisecond * 50 ).
3131 WithMaxConsumeDuration (0 ).
3232 WithLogger (log .New (os .Stderr , "[DelayQueue]" , log .LstdFlags )).
33- WithFetchLimit (2 ).
34- WithNackRedeliveryDelay (time .Second )
33+ WithFetchLimit (2 )
3534
3635 for i := 0 ; i < size ; i ++ {
3736 err := queue .SendDelayMsg (strconv .Itoa (i ), 0 , WithRetryCount (retryCount ), WithMsgTTL (time .Hour ))
@@ -280,7 +279,6 @@ func TestDelayQueue_Massive_Backlog(t *testing.T) {
280279 }
281280}
282281
283-
284282// consume should stopped after actual fetch count hits fetch limit
285283func TestDelayQueue_FetchLimit (t * testing.T ) {
286284 redisCli := redis .NewClient (& redis.Options {
@@ -325,7 +323,7 @@ func TestDelayQueue_FetchLimit(t *testing.T) {
325323 if len (ids2 ) > 0 {
326324 t .Error ("should get 0 message, after hitting fetch limit" )
327325 }
328-
326+
329327 // consume
330328 for _ , id := range ids1 {
331329 queue .callback (id )
@@ -391,4 +389,59 @@ func TestDelayQueue_ScriptPreload(t *testing.T) {
391389 t .Errorf ("expect 1 delivery, actual %d. key: %s" , v , k )
392390 }
393391 }
394- }
392+ }
393+
394+ func TestDelayQueue_NackRedeliveryDelay (t * testing.T ) {
395+ redisCli := redis .NewClient (& redis.Options {
396+ Addr : "127.0.0.1:6379" ,
397+ })
398+ redisCli .FlushDB (context .Background ())
399+ cb := func (s string ) bool {
400+ return false
401+ }
402+ redeliveryDelay := time .Second
403+ queue := NewQueue ("test" , redisCli , UseHashTagKey ()).
404+ WithCallback (cb ).
405+ WithFetchInterval (time .Millisecond * 50 ).
406+ WithLogger (log .New (os .Stderr , "[DelayQueue]" , log .LstdFlags )).
407+ WithDefaultRetryCount (3 ).
408+ WithNackRedeliveryDelay (redeliveryDelay )
409+
410+ err := queue .SendScheduleMsg ("foobar" , time .Now ().Add (- time .Minute ))
411+ if err != nil {
412+ t .Error (err )
413+ }
414+ // first consume, callback will failed
415+ ids , err := queue .beforeConsume ()
416+ if err != nil {
417+ t .Errorf ("consume error: %v" , err )
418+ return
419+ }
420+ for _ , id := range ids {
421+ queue .callback (id )
422+ }
423+ queue .afterConsume ()
424+
425+ // retry immediately
426+ ids , err = queue .beforeConsume ()
427+ if err != nil {
428+ t .Errorf ("consume error: %v" , err )
429+ return
430+ }
431+ if len (ids ) != 0 {
432+ t .Errorf ("should not redeliver immediately" )
433+ return
434+ }
435+
436+ time .Sleep (redeliveryDelay )
437+ queue .afterConsume ()
438+ ids , err = queue .beforeConsume ()
439+ if err != nil {
440+ t .Errorf ("consume error: %v" , err )
441+ return
442+ }
443+ if len (ids ) != 1 {
444+ t .Errorf ("should not redeliver immediately" )
445+ return
446+ }
447+ }
0 commit comments