workermanager

package
v0.9.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 12, 2025 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

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"`
	MaxAttemptRetry   int           `mapstructure:"max_attempt_retry" default:"3"`
}

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
	SoftDeleteByURN(ctx context.Context, params asset.SoftDeleteAssetParams) error
	DeleteByQueryExpr(ctx context.Context, queryExpr queryexpr.ExprStr) error
	DeleteByIsDeletedAndServicesAndUpdatedAt(ctx context.Context, isDeleted bool, services []string, expiryThreshold time.Time) error
	SoftDeleteAssets(ctx context.Context, assets []asset.Asset, doUpdateVersion bool) 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) EnqueueDeleteAssetsByIsDeletedAndServicesAndUpdatedAtJob added in v0.9.1 

func (m *InSituWorker) EnqueueDeleteAssetsByIsDeletedAndServicesAndUpdatedAtJob(
	ctx context.Context,
	isDeleted bool,
	services []string,
	expiryThreshold time.Time,
) error

func (*InSituWorker) EnqueueDeleteAssetsByQueryExprJob added in v0.7.8 

func (m *InSituWorker) EnqueueDeleteAssetsByQueryExprJob(ctx context.Context, queryExpr string) error

func (*InSituWorker) EnqueueIndexAssetJob 

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

func (*InSituWorker) EnqueueSoftDeleteAssetJob added in v0.9.0 

func (m *InSituWorker) EnqueueSoftDeleteAssetJob(ctx context.Context, params asset.SoftDeleteAssetParams) error

func (*InSituWorker) EnqueueSoftDeleteAssetsJob added in v0.9.0 

func (m *InSituWorker) EnqueueSoftDeleteAssetsJob(ctx context.Context, assets []asset.Asset) error

func (*InSituWorker) EnqueueSyncAssetJob added in v0.7.0 

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) DeleteAssetsByQueryExpr added in v0.7.8 

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

func (*Manager) DeleteAssetsByServicesAndUpdatedAt added in v0.9.1 

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

func (*Manager) EnqueueDeleteAssetJob 

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

func (*Manager) EnqueueDeleteAssetsByIsDeletedAndServicesAndUpdatedAtJob added in v0.9.1 

func (m *Manager) EnqueueDeleteAssetsByIsDeletedAndServicesAndUpdatedAtJob(
	ctx context.Context,
	isDeleted bool,
	services []string,
	expiryThreshold time.Time,
) error

func (*Manager) EnqueueDeleteAssetsByQueryExprJob added in v0.7.8 

func (m *Manager) EnqueueDeleteAssetsByQueryExprJob(ctx context.Context, queryExpr string) error

func (*Manager) EnqueueIndexAssetJob 

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

func (*Manager) EnqueueSoftDeleteAssetJob added in v0.9.0 

func (m *Manager) EnqueueSoftDeleteAssetJob(ctx context.Context, softDeleteAsset asset.SoftDeleteAssetParams) error

func (*Manager) EnqueueSoftDeleteAssetsJob added in v0.9.0 

func (m *Manager) EnqueueSoftDeleteAssetsJob(ctx context.Context, assets []asset.Asset) error

func (*Manager) EnqueueSyncAssetJob added in v0.7.0 

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) SoftDeleteAsset added in v0.9.0 

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

func (*Manager) SoftDeleteAssets added in v0.9.0 

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

func (*Manager) SyncAssets added in v0.7.0 

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)
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.