Function Test_Client_StopAndCancel

client_test.go:2442–2539

(t *testing.T)

}

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

	ctx := context.Background()

	type JobArgs struct {
		testutil.JobArgsReflectKind[JobArgs]
	}

	type testBundle struct {
		jobDoneChan    chan struct{}
		jobStartedChan chan int64
	}

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

		config := newTestConfig(t, "")

		var (
			jobStartedChan = make(chan int64)
			jobDoneChan    = make(chan struct{})
		)
		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			jobStartedChan <- job.ID
			t.Logf("Job waiting for context cancellation")
			defer t.Logf("Job finished")
			<-ctx.Done()
			t.Logf("Job context done, closing chan and returning")
			close(jobDoneChan)
			return nil
		}))

		client := runNewTestClient(ctx, t, config)

		return client, &testBundle{
			jobDoneChan:    jobDoneChan,
			jobStartedChan: jobStartedChan,
		}
	}

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

		client, bundle := setup(t)

		startClient(ctx, t, client)

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

		riversharedtest.WaitOrTimeout(t, bundle.jobStartedChan)

		require.NoError(t, client.StopAndCancel(ctx))
		riversharedtest.WaitOrTimeout(t, client.Stopped())
	})

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