package process import ( "context" "errors" "testing" "time" ) // 1190/AC4: given the consume loop, when its context ends, then it stops pulling and returns without // acknowledging a message in hand. func TestRunQueueStopsPullingAndNeverActsOnAMessageWhenContextEnds(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) acked := false msg := &Message{Subject: "test", ack: func() error { acked = true; return nil }, nak: func(time.Duration) error { return nil }, term: func() error { return nil }} pulls := 0 next := func(ctx context.Context) (*Message, error) { pulls++ return msg, nil } handled := 0 done := make(chan error, 1) go func() { done <- runQueue(ctx, next, func(m *Message) { handled++ // The handler does not act on the message at all: a redelivery, a termination or an // acknowledgement is its choice alone, never the loop's. cancel() }) }() select { case err := <-done: if !errors.Is(err, context.Canceled) { t.Fatalf("runQueue returned %v, want context.Canceled", err) } case <-time.After(3 * time.Second): t.Fatal("runQueue did not stop when its context ended") } if handled != 1 { t.Fatalf("handle was called %d times, want 1", handled) } if acked { t.Error("runQueue acknowledged the message itself; only the handler may") } if pulls != 1 { t.Fatalf("next was called %d times, want exactly 1: once the handler ends the context, runQueue's "+ "own check stops it before pulling again", pulls) } } // 1190/AC4: a context already ended before the loop starts pulls nothing at all. func TestRunQueueNeverPullsOnAnAlreadyEndedContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 err := runQueue(ctx, func(context.Context) (*Message, error) { calls++ return nil, context.Canceled }, func(*Message) { t.Fatal("handle was called on an already-ended context") }) if !errors.Is(err, context.Canceled) { t.Fatalf("runQueue returned %v, want context.Canceled", err) } if calls != 0 { t.Fatalf("next was called %d times, want 0", calls) } }