broker

type MessageBroker interface {
	// CreateQueue creates the specified queue if it does not exist or
	// returns the existing queue with the name specified in the
	// QueueDefinition. If a dead letter queue has been specified for
	// the queue, it will be returned as the second queue in returned
	// queue slice.
	CreateQueue(qd *QueueDefinition) ([]Queue, error)
}

type MessageConsumer interface {
	// HandleMessage processes the message. If true is returned, the
	// message is acknowledged and removed from the message queue. If false
	// is returned, this message will reappear on the message queue.
	Consume(qm *QueueManager, q Queue, msg ReceiveMessage) bool
}

type MessageJSON struct {
	GroupID *string
	V       interface{}
}

type OnQueueCreation func(qm *QueueManager, q Queue, name QueueName)

type Queue interface {
	SendOne(SendMessage) error
	// SendMany sends multiple message to the queue.
	SendMany([]SendMessage) error
	// Receive a message from the queue. Don't forget to call the Message's
	// Ack() method once the message has been handled!
	ReceiveOne() (ReceiveMessage, error)
	// Put this ReceiveMessage back on the queue
	Requeue(ReceiveMessage) error
	// Fetch metadata by name from broker.
	FetchAttributes([]string) (map[string]string, error)
	// Remove this queue from the broker. (For testing CreateQueue)
	Delete() error
	// Pretty print the name of the queue
	String() string
	// MoveMessages moves the contents of one queue to another. It returns
	// the number of messages moved.
	MoveMessages(Queue) (int, error)
	// GetApproximateNumberOfMessages
	GetApproximateNumberOfMessages() (*int, error)
}

type QueueDefinition struct {
	// The name of the queue minus the prefix. This is how we refer
	// to the queue in the code and from the command line.
	Name QueueName
	// The name of the queue as it exists on the message broker. This
	// is the name plus a prefix.
	CanonicalName string
	// Queue configuration options
	Options map[string]string
	// Queue (redrive) policy options
	Policy map[string]string
	// Tags are key value pairs that are not used internally, rather they're
	// meant to allow consumers to easily annotate for classification or configuration
	Tags map[string]string

	// If specified, this is the name of the dead letter queue as it exists
	// on the broker. If set, this queue will be created and configured as
	// a dead letter queue for this queue. In addition, any policy options
	// in Policy will be applied to the queue during queue creation.
	DLQ *QueueDefinition
	// if enabled the contents of the message will be compressed while
	// stored on the queue. Message will still be received uncompressed.
	UseGzip bool
	// contains filtered or unexported fields
}

type QueueDefinitionOption func(*QueueDefinition)

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

type QueueManagerOption func(*QueueManager)

type QueueName string

type QueueProcessingOptions struct {
	// amount of parallelism desired (number of consumer go routines)
	Parallelism int
	// consumer of broker library (controller) will pass a message to stop
	Stop chan bool
	// signal to controller that it's safe to exit
	Done chan bool
	// if true, will requeue this message on SIGTERM
	RequeueOnSigterm bool
}

type ReceiveMessage interface {
	// Deletes the from the queue. This *must be done* or the
	// the message will reappear on the queue.
	Ack() error
	// Payload returns the message as it was received into v
	Unmarshal(v interface{}) error
	// Returns the unique identifier representing the message.
	ID() string
	// Sets the number of seconds this message should be invisible to other consumers
	// This overrides the default visibility set on the queue
	SetVisibilityTimeout(n int) error
}

type SendMessage interface {
	// Marshal serializes the payload for transport to the message queue.
	Marshal() ([]byte, error)

	// GroupId returns the group that this message belongs to
	GroupId() *string
}