processor

Documentation

Overview

Package processor provides a processor for the application.

Index

• type Config
  • func DefaultConfig() Config
• type Embedding
  • func NewEmbedding(embeddingProvider embedding.Provider, logger *logrus.Logger) *Embedding
  • func (e *Embedding) Process(ctx context.Context, d *types.Document) (*types.Document, error)
• type Manager
  • func NewManager(storage storage.Storage, registry *Registry, config Config, ...) *Manager
  • func (m *Manager) Start(ctx context.Context)
• type ManagerMetrics
• type Processor
  • func NewProcessor(config Config, storage storage.Storage, logger *logrus.Logger, ...) (*Processor, error)
  • func (p *Processor) Start(ctx context.Context) error
  • func (p *Processor) Stop()
• type ProcessorMetrics
• type ProcessorUnit
• type Registry
  • func NewRegistry(storage storage.Storage, logger *logrus.Logger) *Registry
  • func (r *Registry) Get(name string) (ProcessorUnit, bool)
  • func (r *Registry) GetAll() []ProcessorUnit
  • func (r *Registry) RefreshProcessors(ctx context.Context) error
  • func (r *Registry) Register(name string, processor ProcessorUnit) error
• type Worker
  • func NewWorker(id int, storage storage.Storage, registry *Registry, logger *logrus.Logger, ...) (*Worker, error)
  • func (w *Worker) ProcessDocument(ctx context.Context, docUUID uuid.UUID) error
• type WorkerMetrics

Types

type Config

type Config struct {
	WorkerCount              int
	BatchSize                int
	ProcessInterval          time.Duration
	RetryAttempts            int
	RetryDelay               time.Duration
	ShutdownTimeout          time.Duration
	ProcessorRefreshInterval time.Duration
}

Config contains configuration options for the processor

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a default configuration for the processor

type Embedding

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

Embedding is a processor that adds embeddings to a document

func NewEmbedding

func NewEmbedding(embeddingProvider embedding.Provider, logger *logrus.Logger) *Embedding

NewEmbedding creates a new embedding processor

func (*Embedding) Process

func (e *Embedding) Process(ctx context.Context, d *types.Document) (*types.Document, error)

Process adds embeddings to a document

type Manager

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

Manager is a processor that manages workers to process documents

func NewManager

func NewManager(storage storage.Storage, registry *Registry, config Config, logger *logrus.Logger, meter metric.Meter) *Manager

NewManager creates a new manager with the given settings

func (*Manager) Start

func (m *Manager) Start(ctx context.Context)

Start starts the manager

type ManagerMetrics

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

ManagerMetrics contains metrics for the manager

type Processor

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

Processor is a processor that processes documents

func NewProcessor

func NewProcessor(config Config, storage storage.Storage, logger *logrus.Logger, meter metric.Meter) (*Processor, error)

NewProcessor creates a new processor with the given settings

func (*Processor) Start

func (p *Processor) Start(ctx context.Context) error

Start starts the processor

func (*Processor) Stop

func (p *Processor) Stop()

Stop stops the processor

type ProcessorMetrics

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

ProcessorMetrics contains metrics for the processor

type ProcessorUnit

type ProcessorUnit interface {
	Process(ctx context.Context, d *types.Document) error
}

ProcessorUnit is a processor unit that processes documents

type Registry

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

Registry is a registry of processors

func NewRegistry

func NewRegistry(storage storage.Storage, logger *logrus.Logger) *Registry

NewRegistry creates a new registry

func (*Registry) Get

func (r *Registry) Get(name string) (ProcessorUnit, bool)

Get returns a processor

func (*Registry) GetAll

func (r *Registry) GetAll() []ProcessorUnit

GetAll returns all processors

func (*Registry) RefreshProcessors

func (r *Registry) RefreshProcessors(ctx context.Context) error

RefreshProcessors refreshes the processors in the registry

func (*Registry) Register

func (r *Registry) Register(name string, processor ProcessorUnit) error

Register registers a processor

type Worker

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

Worker is a processing worker that processes documents

func NewWorker

func NewWorker(id int, storage storage.Storage, registry *Registry, logger *logrus.Logger, config Config, meter metric.Meter) (*Worker, error)

NewWorker creates a new worker with the given settings

func (*Worker) ProcessDocument

func (w *Worker) ProcessDocument(ctx context.Context, docUUID uuid.UUID) error

ProcessDocument processes a document

type WorkerMetrics

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

WorkerMetrics contains metrics for the worker