processor

Documentation

Index

• Variables
• type Delayer
• type Processor
  • func New(q Queuer, opt *msgqueue.Options) *Processor
  • func Start(q Queuer, opt *msgqueue.Options) *Processor
  • func (p *Processor) Add(msg *msgqueue.Message) error
  • func (p *Processor) AddDelay(msg *msgqueue.Message, delay time.Duration) error
  • func (p *Processor) Process(msg *msgqueue.Message) error
  • func (p *Processor) ProcessAll() error
  • func (p *Processor) ProcessOne() error
  • func (p *Processor) Purge() error
  • func (p *Processor) Start() error
  • func (p *Processor) Stats() *Stats
  • func (p *Processor) Stop() error
  • func (p *Processor) StopTimeout(timeout time.Duration) error
  • func (p *Processor) String() string
• type Queuer
• type Stats

Variables

var ErrNotSupported = errors.New("processor: not supported")

Types

type Delayer

type Delayer interface {
	Delay() time.Duration
}

type Processor

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

Processor reserves messages from the queue, processes them, and then either releases or deletes messages from the queue.

func New

func New(q Queuer, opt *msgqueue.Options) *Processor

New creates new Processor for the queue using provided processing options.

func Start

func Start(q Queuer, opt *msgqueue.Options) *Processor

Starts creates new Processor and starts it.

func (*Processor) Add

func (p *Processor) Add(msg *msgqueue.Message) error

Add adds message to the processor internal queue.

func (*Processor) AddDelay

func (p *Processor) AddDelay(msg *msgqueue.Message, delay time.Duration) error

Add adds message to the processor internal queue with specified delay.

func (*Processor) Process

func (p *Processor) Process(msg *msgqueue.Message) error

Process is low-level API to process message bypassing the internal queue.

func (*Processor) ProcessAll

func (p *Processor) ProcessAll() error

ProcessAll starts workers to process messages in the queue and then stops them when all messages are processed.

func (*Processor) ProcessOne

func (p *Processor) ProcessOne() error

ProcessOne processes at most one message in the queue.

func (*Processor) Purge

func (p *Processor) Purge() error

Purge discards messages from the internal queue.

func (*Processor) Start

func (p *Processor) Start() error

Start starts processing messages in the queue.

func (*Processor) Stats

func (p *Processor) Stats() *Stats

Stats returns processor stats.

func (*Processor) Stop

func (p *Processor) Stop() error

Stop is StopTimeout with 30 seconds timeout.

func (*Processor) StopTimeout

func (p *Processor) StopTimeout(timeout time.Duration) error

StopTimeout waits workers for timeout duration to finish processing current messages and stops workers.

func (*Processor) String

func (p *Processor) String() string

type Queuer

type Queuer interface {
	Name() string
	Processor() *Processor
	Add(msg *msgqueue.Message) error
	Call(args ...interface{}) error
	CallOnce(dur time.Duration, args ...interface{}) error
	ReserveN(n int) ([]msgqueue.Message, error)
	Release(*msgqueue.Message, time.Duration) error
	Delete(msg *msgqueue.Message) error
	DeleteBatch(msg []*msgqueue.Message) error
	Purge() error
	Close() error
	CloseTimeout(time.Duration) error
}

type Stats

type Stats struct {
	InFlight    uint32
	Deleting    uint32
	Processed   uint32
	Retries     uint32
	Fails       uint32
	AvgDuration time.Duration
}