processor

Documentation

Index

• type ProcessorConfig
  • func DefaultProcessorConfig() *ProcessorConfig
• type QueueProcessor
  • func NewQueueProcessor(cfg *ProcessorConfig, redisClient *redis.Client, ...) *QueueProcessor
  • func (qp *QueueProcessor) GetQueueLength(ctx context.Context) (int64, error)
  • func (qp *QueueProcessor) Start(ctx context.Context) error
• type TransactionProcessor

Types

type ProcessorConfig

type ProcessorConfig struct {
	// Queue name in Redis to process
	QueueName string

	// Maximum number of concurrent workers
	Concurrency int

	// Maximum number of transactions to process in each batch
	BatchSize int64

	// Sleep duration when queue is empty
	EmptyQueueSleep time.Duration
}

ProcessorConfig holds configuration for the queue processor

func DefaultProcessorConfig

func DefaultProcessorConfig() *ProcessorConfig

DefaultProcessorConfig returns a default processor configuration

type QueueProcessor

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

QueueProcessor processes transactions from a Redis queue

func NewQueueProcessor

func NewQueueProcessor(cfg *ProcessorConfig, redisClient *redis.Client, processor TransactionProcessor) *QueueProcessor

NewQueueProcessor creates a new queue processor

func (*QueueProcessor) GetQueueLength

func (qp *QueueProcessor) GetQueueLength(ctx context.Context) (int64, error)

GetQueueLength returns the current length of the processing queue

func (*QueueProcessor) Start

func (qp *QueueProcessor) Start(ctx context.Context) error

Start begins processing the queue and blocks until context is cancelled

type TransactionProcessor

type TransactionProcessor interface {
	// ProcessTransaction processes a single transaction by its ID
	// Returns a list of topics/tokens that this transaction belongs to
	ProcessTransaction(ctx context.Context, txid *chainhash.Hash) ([]string, error)
}

TransactionProcessor defines the interface for processing individual transactions