queue

Documentation

Overview

Package queue implements all the queue handling with tsuru. It abstracts which queue server is being used, how the message gets marshaled in to the wire and how it's read.

It provides a basic type: Message. You can Put, Get, Delete and Release messages, using methods and functions with respective names.

It also provides a generic, thread safe, handler for messages, with start and stop capability.

Index

• func Preempt()
• func Register(name string, factory QFactory)
• type Handler
• type Message
  • func (m *Message) Fail()
• type PubSubQ
• type Q
• type QFactory
  • func Factory() (QFactory, error)

Functions

func Preempt

func Preempt()

Preempt calls Stop and Wait for each running handler.

func Register

func Register(name string, factory QFactory)

Register registers a new queue factory. This is how one would add a new queue to tsuru.

Types

type Handler

type Handler interface {
	// Start starts the handler. It must be safe to call this function
	// multiple times, even if the handler is already running.
	Start()

	// Stop sends a signal to stop the handler, it won't stop the handler
	// immediately. After calling Stop, one should call Wait for blocking
	// until the handler is stopped.
	//
	// This method will return an error if the handler is not running.
	Stop() error

	// Wait blocks until the handler actually stops.
	Wait()
}

Handler represents a runnable routine. It can be started and stopped.

type Message

type Message struct {
	Action string
	Args   []string
	// contains filtered or unexported fields
}

Message represents the message stored in the queue.

A message is specified by an action and a slice of strings, representing arguments to the action.

For example, the action "regenerate apprc" could receive one argument: the name of the app for which the apprc file will be regenerate.

func (*Message) Fail

func (m *Message) Fail()

Fail marks the message as failed, telling the handler to requeue it.

type PubSubQ

type PubSubQ interface {
	Q

	// Publishes a message using the underlaying queue server.
	Pub(msg []byte) error

	// Returns a channel that will yield every message published to this
	// queue.
	Sub() (chan []byte, error)

	// Unsubscribe the queue, this should make sure the channel returned
	// by Sub() is closed.
	UnSub() error
}

PubSubQ represents a Q implementation that allows Publishing and Subscribing messages.

type Q

type Q interface {
	// Get retrieves a message from the queue.
	Get(timeout time.Duration) (*Message, error)

	// Put sends a message to the queue after the given delay. When delay
	// is 0, the message is sent immediately to the queue.
	Put(m *Message, delay time.Duration) error
}

Q represents a queue. A queue is a type that supports the set of operations described by this interface.

type QFactory

type QFactory interface {
	// Get returns a queue instance, identified by the given name.
	Get(name string) (Q, error)

	// Handler returns a handler for the given queue names. Once the
	// handler is started (after calling Start method), it will call f
	// whenever a new message arrives in one of the given queue names.
	Handler(f func(*Message), name ...string) (Handler, error)
}

QFactory manages queues. It's able to create new queue and handler instances.

func Factory

func Factory() (QFactory, error)

Factory returns an instance of the QFactory used in tsuru. It reads tsuru configuration to find the currently used queue system (for example, beanstalkd) and returns an instance of the configured system, if it's registered. Otherwise it will return an error.