Function TestNotifier

internal/notifier/notifier_test.go:32–625

(t *testing.T)

}

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

	const (
		testTopic1 = "test_topic1"
		testTopic2 = "test_topic2"
	)

	ctx := context.Background()

	type testBundle struct {
		dbPool *pgxpool.Pool
		exec   riverdriver.Executor
		schema string
	}

	type testOpts struct {
		dbPool *pgxpool.Pool // may be left nil for riversharedtest.DBPool
	}

	setup := func(t *testing.T, opts *testOpts) (*Notifier, *testBundle) {
		t.Helper()

		if opts == nil {
			opts = &testOpts{}
		}

		var (
			dbPool   = cmp.Or(opts.dbPool, riversharedtest.DBPool(ctx, t))
			driver   = riverpgxv5.New(dbPool)
			schema   = riverdbtest.TestSchema(ctx, t, driver, nil)
			listener = driver.GetListener(&riverdriver.GetListenenerParams{Schema: schema})
		)

		notifier := New(riversharedtest.BaseServiceArchetype(t), listener)
		notifier.testSignals.Init(t)

		return notifier, &testBundle{
			dbPool: dbPool,
			exec:   driver.GetExecutor(),
			schema: schema,
		}
	}

	start := func(t *testing.T, notifier *Notifier) {
		t.Helper()

		require.NoError(t, notifier.Start(ctx))
		t.Cleanup(notifier.Stop)
	}

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

		notifier, _ := setup(t, nil)
		start(t, notifier)

		notifier.testSignals.ListeningBegin.WaitOrTimeout()