Function TestTaskQueue

pkg/redis/redis_test.go:364–488

(t *testing.T)

}

func TestTaskQueue(t *testing.T) {
	a, ctx := test.New(t)

	cl, flush := test.NewRedis(ctx, "redis_test")
	defer flush()
	defer cl.Close()

	q := &TaskQueue{
		Redis:            cl,
		MaxLen:           16384,
		Group:            "testGroup",
		Key:              cl.Key("test"),
		StreamBlockLimit: testStreamBlockLimit(),
	}

	err := q.Init(ctx)
	a.So(err, should.BeNil)
	defer func() {
		err := q.Close(ctx)
		a.So(err, should.BeNil)
	}()

	errCh := make(chan error, 1)
	cancelCtx, cancel := context.WithCancel(ctx)
	go func() {
		errCh <- q.Dispatch(cancelCtx, "testID", nil)
	}()
	defer func() {
		cancel()

		select {
		case <-ctx.Done():
			t.Error("Timed out while waiting for Dispatch to finish running")
		case err := <-errCh:
			if !a.So(errors.IsCanceled(err), should.BeTrue) {
				t.Errorf("DispatchTask failed with: %s", test.FormatError(err))
			}
		}
	}()

	assertPop := func(ctx context.Context, expectedPayload string, expectedStartAt time.Time) bool {
		t, a := test.MustNewTFromContext(ctx)
		t.Helper()

		type popFuncReq struct {
			Pipeliner redis.Pipeliner
			Payload   string
			Time      time.Time
			Response  chan<- error
		}

		var called bool
		errCh := make(chan error, 1)
		go func() {
			errCh <- q.Pop(ctx, "testID", nil, func(p redis.Pipeliner, payload string, startAt time.Time) error {
				p.Ping(ctx)
				a.So(called, should.BeFalse)
				a.So(payload, should.Equal, expectedPayload)