workermanager

Documentation

Index

• type Config
• type Deps
• type DiscoveryRepository
• type InSituWorker
  • func NewInSituWorker(deps Deps) *InSituWorker
  • func (*InSituWorker) Close() error
  • func (m *InSituWorker) EnqueueDeleteAssetJob(ctx context.Context, urn string) error
  • func (m *InSituWorker) EnqueueIndexAssetJob(ctx context.Context, ast asset.Asset) error
• type Manager
  • func New(ctx context.Context, deps Deps) (*Manager, error)
  • func NewWithWorker(w Worker, deps Deps) *Manager
  • func (m *Manager) Close() error
  • func (m *Manager) DeleteAsset(ctx context.Context, job worker.JobSpec) error
  • func (m *Manager) EnqueueDeleteAssetJob(ctx context.Context, urn string) error
  • func (m *Manager) EnqueueIndexAssetJob(ctx context.Context, ast asset.Asset) error
  • func (m *Manager) IndexAsset(ctx context.Context, job worker.JobSpec) error
  • func (m *Manager) Run(ctx context.Context) error
• type Worker

Types

type Config

type Config struct {
	Enabled           bool          `mapstructure:"enabled"`
	WorkerCount       int           `mapstructure:"worker_count" default:"3"`
	PollInterval      time.Duration `mapstructure:"poll_interval" default:"500ms"`
	ActivePollPercent float64       `mapstructure:"active_poll_percent" default:"20"`
	PGQ               pgq.Config    `mapstructure:"pgq"`
	JobManagerPort    int           `mapstructure:"job_manager_port"`
}

type Deps

type Deps struct {
	Config        Config
	DiscoveryRepo DiscoveryRepository
	Logger        log.Logger
}

type DiscoveryRepository

type DiscoveryRepository interface {
	Upsert(context.Context, asset.Asset) error
	DeleteByURN(ctx context.Context, assetURN string) error
}

type InSituWorker

type InSituWorker struct {
	// contains filtered or unexported fields
}

func NewInSituWorker

func NewInSituWorker(deps Deps) *InSituWorker

func (*InSituWorker) Close

func (*InSituWorker) Close() error

func (*InSituWorker) EnqueueDeleteAssetJob

func (m *InSituWorker) EnqueueDeleteAssetJob(ctx context.Context, urn string) error

func (*InSituWorker) EnqueueIndexAssetJob

func (m *InSituWorker) EnqueueIndexAssetJob(ctx context.Context, ast asset.Asset) error

type Manager

type Manager struct {
	// contains filtered or unexported fields
}

func New

func New(ctx context.Context, deps Deps) (*Manager, error)

func NewWithWorker

func NewWithWorker(w Worker, deps Deps) *Manager

func (*Manager) Close

func (m *Manager) Close() error

func (*Manager) DeleteAsset

func (m *Manager) DeleteAsset(ctx context.Context, job worker.JobSpec) error

func (*Manager) EnqueueDeleteAssetJob

func (m *Manager) EnqueueDeleteAssetJob(ctx context.Context, urn string) error

func (*Manager) EnqueueIndexAssetJob

func (m *Manager) EnqueueIndexAssetJob(ctx context.Context, ast asset.Asset) error

func (*Manager) IndexAsset

func (m *Manager) IndexAsset(ctx context.Context, job worker.JobSpec) error

func (*Manager) Run

func (m *Manager) Run(ctx context.Context) error

type Worker

type Worker interface {
	Register(typ string, h worker.JobHandler) error
	Run(ctx context.Context) error
	Enqueue(ctx context.Context, jobs ...worker.JobSpec) error
}