opqueue

Documentation

Index

• type Config
• type OperationMessage
• type Queue
  • func New(cfg Config, pubSub pubSub, p storage.Provider, taskMgr taskManager, ...) (*Queue, error)
  • func (q *Queue) Add(op *operation.QueuedOperation, protocolVersion uint64) (uint, error)
  • func (q *Queue) Len() uint
  • func (q *Queue) Peek(num uint) (operation.QueuedOperationsAtTime, error)
  • func (q *Queue) Remove(num uint) (ops operation.QueuedOperationsAtTime, ack func() uint, nack func(), err error)

Types

type Config

type Config struct {
	// PoolSize is the number of AMQP subscribers that are listening for operation messages.
	PoolSize int
	// TaskMonitorInterval is the interval (period) in which operation queue tasks from other server instances
	// are monitored.
	TaskMonitorInterval time.Duration
	// TaskExpiration is the maximum time that an operation queue task can exist in the database before it is
	// considered to have expired. At which point, any other server instance may delete the task and take over
	// processing of all operations associated with the task.
	TaskExpiration time.Duration
	// MaxRetries is the maximum number of retries for a failed operation in a batch.
	MaxRetries int
	// RetriesInitialDelay is the delay for the initial retry attempt.
	RetriesInitialDelay time.Duration
	// RetriesMaxDelay is the maximum delay for a retry attempt.
	RetriesMaxDelay time.Duration
	// RetriesMultiplier is the multiplier for a retry attempt. For example, if set to 1.5 and
	// the previous retry interval was 2s then the next retry interval is set 3s.
	RetriesMultiplier float64
}

Config contains configuration parameters for the operation queue.

type OperationMessage

type OperationMessage struct {
	ID        string                           `json:"id"`
	Operation *operation.QueuedOperationAtTime `json:"operation"`
	Retries   int                              `json:"retries"`
}

OperationMessage contains the data that is sent to the message broker.

type Queue

type Queue struct {
	*lifecycle.Lifecycle
	// contains filtered or unexported fields
}

Queue implements an operation queue that uses a publisher/subscriber.

func New

func New(cfg Config, pubSub pubSub, p storage.Provider, taskMgr taskManager, metrics metricsProvider) (*Queue, error)

New returns a new operation queue.

func (*Queue) Add

func (q *Queue) Add(op *operation.QueuedOperation, protocolVersion uint64) (uint, error)

Add publishes the given operation.

func (*Queue) Len

func (q *Queue) Len() uint

Len returns the length of the pending queue.

func (*Queue) Peek

func (q *Queue) Peek(num uint) (operation.QueuedOperationsAtTime, error)

Peek returns (up to) the given number of operations from the head of the queue but does not remove them.

func (*Queue) Remove

func (q *Queue) Remove(num uint) (ops operation.QueuedOperationsAtTime, ack func() uint, nack func(), err error)

Remove removes (up to) the given number of items from the head of the queue. Returns the actual number of items that were removed and the new length of the queue.