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
  • func (m *InSituWorker) EnqueueSyncAssetJob(ctx context.Context, service string) 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) EnqueueSyncAssetJob(ctx context.Context, service string) error
  • func (m *Manager) IndexAsset(ctx context.Context, job worker.JobSpec) error
  • func (m *Manager) Run(ctx context.Context) error
  • func (m *Manager) SyncAssets(ctx context.Context, job worker.JobSpec) 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"`
	SyncJobTimeout    time.Duration `mapstructure:"sync_job_timeout" default:"15m"`
	IndexJobTimeout   time.Duration `mapstructure:"index_job_timeout" default:"5s"`
	DeleteJobTimeout  time.Duration `mapstructure:"delete_job_timeout" default:"5s"`
}

type Deps

type Deps struct {
	Config        Config
	DiscoveryRepo DiscoveryRepository
	AssetRepo     asset.Repository
	Logger        log.Logger
}

type DiscoveryRepository

type DiscoveryRepository interface {
	Upsert(context.Context, asset.Asset) error
	DeleteByURN(ctx context.Context, assetURN string) error
	SyncAssets(ctx context.Context, indexName string) (cleanupFn func() error, err 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

func (*InSituWorker) EnqueueSyncAssetJob

func (m *InSituWorker) EnqueueSyncAssetJob(ctx context.Context, service string) 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) EnqueueSyncAssetJob

func (m *Manager) EnqueueSyncAssetJob(ctx context.Context, service string) 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

func (*Manager) SyncAssets

func (m *Manager) SyncAssets(ctx context.Context, job worker.JobSpec) 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
	GetSyncJobsByService(ctx context.Context, service string) ([]worker.Job, error)
}