processor

Documentation

Index

• type Config
  • func (c *Config) Validate() error
• type LeaderElectionConfig
• type Manager
  • func NewManager(log logrus.FieldLogger, config *Config, pool *ethereum.Pool, state *s.Manager, ...) (*Manager, error)
  • func (m *Manager) GetQueueName() string
  • func (m *Manager) QueueBlockManually(ctx context.Context, processorName string, blockNumber uint64) (*QueueResult, error)
  • func (m *Manager) Start(ctx context.Context) error
  • func (m *Manager) Stop(ctx context.Context) error
• type QueueResult
• type WorkerConfig

Types

type Config

type Config struct {
	// Processing interval
	Interval time.Duration `yaml:"interval"`

	// Processing mode: forwards, backwards
	Mode string `yaml:"mode"`

	// Maximum concurrent transactions to process
	Concurrency int `yaml:"concurrency"`

	// Leader election configuration
	LeaderElection LeaderElectionConfig `yaml:"leaderElection"`

	// Queue control configuration
	MaxProcessQueueSize    int     `yaml:"maxProcessQueueSize"`
	BackpressureHysteresis float64 `yaml:"backpressureHysteresis"`

	// Processor configurations
	TransactionStructlog structlog.Config `yaml:"transactionStructlog"`
}

Config holds the unified processor configuration

func (*Config) Validate

func (c *Config) Validate() error

type LeaderElectionConfig

type LeaderElectionConfig struct {
	// Enable leader election (default: true)
	Enabled bool `yaml:"enabled"`

	// TTL for leader lock (default: 10s)
	TTL time.Duration `yaml:"ttl"`

	// Renewal interval (default: 3s)
	RenewalInterval time.Duration `yaml:"renewalInterval"`

	// Optional node ID (auto-generated if empty)
	NodeID string `yaml:"nodeId"`
}

LeaderElectionConfig holds configuration for leader election

type Manager

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

Manager coordinates multiple processors with distributed task processing

func NewManager

func NewManager(log logrus.FieldLogger, config *Config, pool *ethereum.Pool, state *s.Manager, redis *r.Client, redisPrefix string) (*Manager, error)

func (*Manager) GetQueueName

func (m *Manager) GetQueueName() string

GetQueueName returns the current queue name based on processing mode

func (*Manager) QueueBlockManually

func (m *Manager) QueueBlockManually(ctx context.Context, processorName string, blockNumber uint64) (*QueueResult, error)

QueueBlockManually allows manual queuing of a specific block for processing

func (*Manager) Start

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

func (*Manager) Stop

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

type QueueResult

type QueueResult struct {
	TransactionCount int
	TasksCreated     int
}

QueueResult contains the result of queuing a block

type WorkerConfig

type WorkerConfig struct {
	Enabled     bool `yaml:"enabled"`
	Concurrency int  `yaml:"concurrency"`
}

WorkerConfig holds worker configuration