Function Test_SubscriptionManager

subscription_manager_test.go:27–199

(t *testing.T)

)

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

	ctx := context.Background()

	type testBundle struct {
		exec        riverdriver.Executor
		subscribeCh chan []jobcompleter.CompleterJobUpdated
		tx          pgx.Tx
	}

	setup := func(t *testing.T) (*subscriptionManager, *testBundle) {
		t.Helper()

		tx := riverdbtest.TestTxPgx(ctx, t)
		exec := riverpgxv5.New(nil).UnwrapExecutor(tx)

		subscribeCh := make(chan []jobcompleter.CompleterJobUpdated, 1)
		manager := newSubscriptionManager(riversharedtest.BaseServiceArchetype(t), subscribeCh)

		require.NoError(t, manager.Start(ctx))
		t.Cleanup(manager.Stop)

		return manager, &testBundle{
			exec:        exec,
			subscribeCh: subscribeCh,
			tx:          tx,
		}
	}

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

		manager, bundle := setup(t)
		t.Cleanup(func() { close(bundle.subscribeCh) })

		sub, cancelSub := manager.SubscribeConfig(&SubscribeConfig{ChanSize: 10, Kinds: []EventKind{EventKindJobCompleted, EventKindJobSnoozed}})
		t.Cleanup(cancelSub)

		// Send some events
		job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), FinalizedAt: ptrutil.Ptr(time.Now())})
		job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCancelled), FinalizedAt: ptrutil.Ptr(time.Now())})
		job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRetryable)})
		job4 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateScheduled)})

		makeStats := func(complete, wait, run time.Duration) *jobstats.JobStatistics {
			return &jobstats.JobStatistics{
				CompleteDuration:  complete,
				QueueWaitDuration: wait,
				RunDuration:       run,
			}
		}

		bundle.subscribeCh <- []jobcompleter.CompleterJobUpdated{
			{Job: job1, JobStats: makeStats(101, 102, 103)}, // completed, should be sent
			{Job: job2, JobStats: makeStats(201, 202, 203)}, // cancelled, should be skipped
		}
		bundle.subscribeCh <- []jobcompleter.CompleterJobUpdated{