queue

Documentation

Index

• type Processor
• type Service
  • func NewService(name string, maxRetries, bufferSize int, ctx context.Context, ...) *Service
  • func (s *Service) Close()
  • func (s *Service) Enqueue(message indexer.Message)
  • func (s *Service) Start(p Processor) error
• type UserOpService
  • func NewUserOpService(db *db.DB, evm indexer.EVMRequester, fb *firebase.PushService) *UserOpService
  • func (s *UserOpService) Process(messages []indexer.Message) (invalid []indexer.Message, errors []error)

Types

type Processor

type Processor interface {
	Process([]indexer.Message) ([]indexer.Message, []error) // Process method to process a message
}

Processor is an interface that must be implemented by the consumer of the queue

type Service

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

Service struct represents a queue service with a queue channel, quit channel, maximum retries, context and a webhook messager.

func NewService

func NewService(name string, maxRetries, bufferSize int, ctx context.Context, wm indexer.WebhookMessager) *Service

NewService function initializes a new Service with provided maximum retries, context and webhook messager.

func (*Service) Close

func (s *Service) Close()

Close method sends a signal to the quit channel to stop the service.

func (*Service) Enqueue

func (s *Service) Enqueue(message indexer.Message)

Enqueue method enqueues a message to the queue channel.

func (*Service) Start

func (s *Service) Start(p Processor) error

Start method starts the service and processes messages from the queue channel. If processing a message fails, it requeues the message until the maximum retries is reached. If the queue was empty, it waits for a duration before continuing to avoid a busy loop. It also notifies errors using the webhook messager. The service can be stopped by sending a signal to the quit channel.

type UserOpService

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

func NewUserOpService

func NewUserOpService(db *db.DB,
	evm indexer.EVMRequester, fb *firebase.PushService) *UserOpService

func (*UserOpService) Process

func (s *UserOpService) Process(messages []indexer.Message) (invalid []indexer.Message, errors []error)

Process method processes messages of type []indexer.Message and returns processed messages and an errors if any.