Function TestProducer_WithNotifier

producer_test.go:217–269

(t *testing.T)

}

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

	testProducer(t, func(ctx context.Context, t *testing.T) (*producer, chan []jobcompleter.CompleterJobUpdated) {
		t.Helper()

		var (
			archetype  = riversharedtest.BaseServiceArchetype(t)
			dbPool     = riversharedtest.DBPool(ctx, t)
			driver     = riverpgxv5.New(dbPool)
			exec       = driver.GetExecutor()
			jobUpdates = make(chan []jobcompleter.CompleterJobUpdated, 10)
			schema     = riverdbtest.TestSchema(ctx, t, driver, nil)
			listener   = driver.GetListener(&riverdriver.GetListenenerParams{Schema: schema})
			pilot      = &riverpilot.StandardPilot{}
			queueName  = fmt.Sprintf("test-producer-with-notifier-%05d", randutil.IntBetween(1, 100_000))
		)

		completer := jobcompleter.NewInlineCompleter(archetype, schema, exec, &riverpilot.StandardPilot{}, jobUpdates)
		{
			require.NoError(t, completer.Start(ctx))
			t.Cleanup(completer.Stop)
		}

		notifier := notifier.New(archetype, listener)
		{
			require.NoError(t, notifier.Start(ctx))
			t.Cleanup(notifier.Stop)
		}

		return newProducer(archetype, exec, pilot, &producerConfig{
			ClientID:                     testClientID,
			Completer:                    completer,
			ErrorHandler:                 newTestErrorHandler(),
			FetchCooldown:                FetchCooldownDefault,
			FetchPollInterval:            50 * time.Millisecond, // more aggressive than normal so in case we miss the event, tests still pass quickly
			HookLookupByJob:              hooklookup.NewJobHookLookup(),
			HookLookupGlobal:             hooklookup.NewHookLookup(nil),
			JobTimeout:                   JobTimeoutDefault,
			MaxWorkers:                   1_000,
			MiddlewareLookupGlobal:       middlewarelookup.NewMiddlewareLookup(nil),
			Notifier:                     notifier,
			Queue:                        queueName,
			QueuePollInterval:            queuePollIntervalDefault,
			QueueReportInterval:          queueReportIntervalDefault,
			RetryPolicy:                  &DefaultClientRetryPolicy{},
			SchedulerInterval:            riverinternaltest.SchedulerShortInterval,
			Schema:                       schema,
			StaleProducerRetentionPeriod: time.Minute,
			Workers:                      NewWorkers(),
		}), jobUpdates
	})
}

func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testing.T) (*producer, chan []jobcompleter.CompleterJobUpdated)) {
	t.Helper()