Function TestResumableStepCursor

resumable_test.go:166–361

(t *testing.T)

}

func TestResumableStepCursor(t *testing.T) {
	t.Parallel()

	type resumableCursor struct {
		ID int `json:"id"`
	}

	setup := func(t *testing.T, metadata string) (context.Context, map[string]any, *rivertype.JobRow) {
		t.Helper()

		metadataUpdates := make(map[string]any)
		ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates)

		return ctx, metadataUpdates, &rivertype.JobRow{Metadata: []byte(metadata)}
	}

	t.Run("DuplicateStepNameSharedWithCursorStep", func(t *testing.T) {
		t.Parallel()

		ctx, _, job := setup(t, `{}`)

		var ran []string
		err := (&rivermiddleware.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error {
			ResumableStep(ctx, "step1", nil, func(ctx context.Context) error {
				ran = append(ran, "step")
				return nil
			})
			ResumableStepCursor(ctx, "step1", nil, func(ctx context.Context, cursor resumableCursor) error {
				ran = append(ran, "cursor")
				return nil
			})

			return nil
		})
		require.EqualError(t, err, `river: duplicate resumable step name "step1"`)
		require.Equal(t, []string{"step"}, ran)
	})

	t.Run("ResumesCursor", func(t *testing.T) {
		t.Parallel()

		ctx, metadataUpdates, job := setup(t, `{}`)

		var (
			cursorResult resumableCursor
			ran          []int
		)
		err := (&rivermiddleware.ResumableMiddleware{}).Work(ctx, job, func(ctx context.Context) error {
			ResumableStep(ctx, "step1", nil, func(ctx context.Context) error {
				ran = append(ran, 1)
				return nil
			})
			ResumableStepCursor(ctx, "step2", nil, func(ctx context.Context, cursor resumableCursor) error {
				cursorResult = cursor
				ran = append(ran, 2)
				require.NoError(t, ResumableSetCursor(ctx, resumableCursor{ID: 42}))
				return errors.New("step2 failed")
			})