Function NewClient

client.go:781–1068

NewClient creates a new Client with the given database driver and configuration. Currently only one driver is supported, which is Pgx v5. See package riverpgxv5. The function takes a generic parameter TTx representing a transaction type, but it can be omitted because it'll generally always be infe

(driver riverdriver.Driver[TTx], config *Config)

//		// handle error
//	}
func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client[TTx], error) {
	if driver == nil {
		return nil, errMissingDriver
	}
	if config == nil {
		return nil, errMissingConfig
	}

	config = config.WithDefaults()

	if err := config.validate(); err != nil {
		return nil, err
	}

	archetype := baseservice.NewArchetype(config.Logger)
	if config.Test.Time != nil {
		if withStub, ok := config.Test.Time.(baseservice.TimeGeneratorWithStub); ok {
			archetype.Time = withStub
		} else {
			archetype.Time = &baseservice.TimeGeneratorWithStubWrapper{TimeGenerator: config.Test.Time}
		}
	}

	for _, hook := range config.Hooks {
		if withBaseService, ok := hook.(baseservice.WithBaseService); ok {
			baseservice.Init(archetype, withBaseService)
		}
	}

	client := &Client[TTx]{
		clientNotifyBundle: &ClientNotifyBundle[TTx]{
			config: config,
			driver: driver,
		},
		config:               config,
		driver:               driver,
		hookLookupByJob:      hooklookup.NewJobHookLookup(),
		hookLookupGlobal:     hooklookup.NewHookLookup(config.Hooks),
		producersByQueueName: make(map[string]*producer),
		testSignals:          clientTestSignals{},
		workCancel:           func(cause error) {}, // replaced on start, but here in case StopAndCancel is called before start up
	}

	client.queues = &QueueBundle{
		clientFetchCooldown:     config.FetchCooldown,
		clientFetchPollInterval: config.FetchPollInterval,
		clientWillExecuteJobs:   config.willExecuteJobs(),
		producerAdd:             client.producerAdd,
		producerRemove:          client.producerRemove,
	}

	baseservice.Init(archetype, &client.baseService)
	client.baseService.Name = "Client" // Have to correct the name because base service isn't embedded like it usually is
	client.insertNotifyLimiter = notifylimiter.NewLimiter(archetype, config.FetchCooldown)

	// Validation ensures that config.JobInsertMiddleware/WorkerMiddleware or
	// the more abstract config.Middleware for middleware are set, but not both,
	// so in practice we never append all three of these to each other.