Function Test_Client_ErrorHandler

client_test.go:5447–5568

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		schema        string
		subscribeChan <-chan *Event
	}

	setup := func(t *testing.T, config *Config) (*Client[pgx.Tx], *testBundle) {
		t.Helper()

		client := runNewTestClient(ctx, t, config)

		subscribeChan, cancel := client.Subscribe(EventKindJobCompleted, EventKindJobFailed)
		t.Cleanup(cancel)

		return client, &testBundle{
			schema:        client.config.Schema,
			subscribeChan: subscribeChan,
		}
	}

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

		config := newTestConfig(t, "")

		type JobArgs struct {
			testutil.JobArgsReflectKind[JobArgs]
		}

		handlerErr := errors.New("job error")
		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			return handlerErr
		}))

		var errorHandlerCalled bool
		config.ErrorHandler = &testErrorHandler{
			HandleErrorFunc: func(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult {
				require.Equal(t, handlerErr, err)
				errorHandlerCalled = true
				return &ErrorHandlerResult{}
			},
		}

		client, bundle := setup(t, config)

		_, err := client.Insert(ctx, JobArgs{}, nil)
		require.NoError(t, err)

		riversharedtest.WaitOrTimeout(t, bundle.subscribeChan)

		require.True(t, errorHandlerCalled)
	})

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