Function TestPopTask

pkg/redis/redis_test.go:157–362

(t *testing.T)

}

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

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

	const (
		testGroup = "testGroup"

		testKey1 = "testKey1"
		testKey2 = "testKey2"
	)

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

		var called bool
		errCh := make(chan error, 1)
		go func() {
			errCh <- PopTask(
				ctx,
				cl.Client,
				testGroup,
				"testID",
				func(p redis.Pipeliner, payload string, startAt time.Time) error {
					p.Ping(ctx)
					if !test.AllTrue(
						a.So(called, should.BeFalse),
						a.So(payload, should.Equal, expectedPayload),
						a.So(startAt, should.Resemble, expectedStartAt),
					) {
						t.Errorf(
							"PopTask assertion failed for task with expected payload %s and expected starting time of %s", //nolint:lll
							expectedPayload,
							expectedStartAt,
						)
					}
					called = true
					return nil
				},
				inputKey,
				testStreamBlockLimit(),
			)
		}()

		select {
		case <-ctx.Done():
			t.Error("Timed out while waiting for Pop callback to be called")
			return false

		case err := <-errCh:
			if !a.So(err, should.BeNil) {
				t.Errorf("PopTask failed with: %s", test.FormatError(err))
			}
			return a.So(called, should.BeTrue) && !a.Failed()
		}