Documentation
¶
Index ¶
Constants ¶
const ( Actions = "Action" DirectQueue = "DirectQueue" MainQueue = "MainQueue" BackupQueue = "BackupQueue" // MaxDeliveryAttempts caps how many times an action is offered to the node before it // is given up, so an action the node can never accept cannot block its queue forever. MaxDeliveryAttempts = 5 )
Variables ¶
var ( // ErrInvalidQueueID is returned when an unrecognized queue ID is provided. ErrInvalidQueueID = errors.New("invalid queue id") // ErrDeliveryExhausted is returned by Delivery.Restore once MaxDeliveryAttempts is spent. ErrDeliveryExhausted = errors.New("delivery attempts exhausted") )
Functions ¶
func DirectInstructionToAction ¶
func DirectInstructionToAction(i *types.DirectInstruction) (*types.Action, error)
DirectInstructionToAction converts a DirectInstruction into an Action with a randomly generated ID.
Types ¶
type ActionQueues ¶
type ActionQueues struct {
// contains filtered or unexported fields
}
ActionQueues manages direct, main, and backup Redis-backed queues for action submission.
func NewActionQueues ¶
func NewActionQueues(client *redis.Client, actionTTL time.Duration, m *metrics.Metrics) *ActionQueues
NewActionQueues creates a new ActionQueues backed by the given Redis client. m may be nil or disabled.
func (*ActionQueues) Dequeue ¶
func (as *ActionQueues) Dequeue(ctx context.Context, queueID processorutils.QueueID) (*Delivery, error)
Dequeue takes the next action off the indicated queue and returns it as a Delivery the caller must Commit or Restore. If no action is available, wrapped ErrEmptyQueue is returned.
ctx cancellation is honoured only before the queue ID is consumed: an interrupted body fetch would leave the action neither queued nor delivered.
func (*ActionQueues) Enqueue ¶
func (as *ActionQueues) Enqueue(ctx context.Context, action *types.Action, queueID processorutils.QueueID) error
Enqueue stores the action and appends its submission ID to the indicated queue.
func (*ActionQueues) QueueLength ¶
func (as *ActionQueues) QueueLength(ctx context.Context) (int64, error)
QueueLength returns the number of elements in the main queue.
type ActionSubmissionID ¶
type ActionSubmissionID struct {
ActionID common.Hash
SubmissionTag types.SubmissionTag
// Attempt counts the deliveries already attempted; absent on entries queued before
// redelivery existed, and never part of the action's storage key.
Attempt uint8 `json:"attempt,omitempty"`
}
ActionSubmissionID uniquely identifies an action by its action ID and submission tag.
func (*ActionSubmissionID) String ¶
func (id *ActionSubmissionID) String() string
String returns a combined string representation of the action ID and submission tag.
type Delivery ¶ added in v0.0.23
type Delivery struct {
// Action is the dequeued action.
Action *types.Action
// contains filtered or unexported fields
}
Delivery is an action taken off its queue whose body is still stored. Exactly one of Commit or Restore must be called once the send outcome is known; calling neither leaves the action off its queue until the body expires.
func (*Delivery) Commit ¶ added in v0.0.23
Commit drops the action body once the node has received the action.
Source Files