README
¶
relay
Relay is a Go framework for task queues, and makes writing code for publishers and consumers very simple. It is a wrapper around the AMQP protocol, and relies on a message broker like RabbitMQ.
The reason Relay exists is that AMQP is a tedious protocol to deal with, and the high level abstraction of a task queue is often something that is desirable. With Relay, you simply Publish objects into task queues, and Consume them on the other end.
Features
- Simple to use, hides the AMQP details
- Flexible encoding and decoding support
- Configuration changes instead of code changes
Documentation
See the online documentation here: http://godoc.org/github.com/armon/relay.
Example
Here is an example of a simple publisher:
conf := &relay.Config{Addr: "rabbitmq"}
conn, err := relay.New(conf)
defer conn.Close()
pub, err := conn.Publisher("tasks")
defer pub.Close()
pub.Publish("this is a test message")
Here is an example of a simple consumer:
conf := &relay.Config{Addr: "rabbitmq"}
conn, err := relay.New(conf)
defer conn.Close()
cons, err := conn.Consumer("tasks")
defer cons.Close()
var msg string
for {
cons.ConsumeAck(&msg)
fmt.Printf("Got msg: %s\n", msg)
}
Here is an example of a consumer using prefetching and multi Acks:
conf := &relay.Config{Addr: "rabbitmq", PrefetchCount: 5, EnableMultiAck: true}
conn, err := relay.New(conf)
defer conn.Close()
cons, err := conn.Consumer("tasks")
defer cons.Close()
var msg string
for {
// Consume 5 messages
for i := 0; i < 5; i++ {
cons.Consume(&msg)
fmt.Printf("Got msg: %s\n", msg)
}
// Acks the last 5 messages
cons.Ack()
}
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ChannelClosed = fmt.Errorf("Channel closed!")
Returned to indicate a closed channel
var TimedOut = fmt.Errorf("Timeout")
Returned to indicate a read timeout
Functions ¶
func IsDecodeFailure ¶
IsDecodeFailure is a helper to determine if the error returned is a deserialization error.
Types ¶
type Config ¶
type Config struct {
Addr string // Host address to dial
Port int // Host por to bind
Vhost string // Broker Vhost
Username string // Broker username
Password string // Broker password
EnableTLS bool // Broker TLS connection
PrefetchCount int // How many messages to prefetch. If < 1, defaults to 1.
EnableMultiAck bool // Controls if we allow multi acks
DisablePublishConfirm bool // Disables confirmations of publish
DisablePersistence bool // Disables message persistence
Exchange string // Custom exchange. Defaults to "relay"
Serializer Serializer // Defaults to GOBSerializer
MessageTTL time.Duration // Optional, attempts to put a TTL on message life
QueueTTL time.Duration // Optional, attempts to make a TTL on a queue life
DelayQueueArgs map[string](map[string]interface{})
}
Config is passed into New when creating a Relay to tune various parameters around broker interactions.
func ConfigFromURI ¶
ConfigFromURI attempts to parse the given AMQP URI according to the spec and return a relay config based on it. See http://www.rabbitmq.com/uri-spec.html.
Default values for the fields are:
Scheme: amqp Host: localhost Port: 5672 Username: guest Password: guest Vhost: /
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer is a type that is used only for consuming messages from a single queue. Multiple Consumers can multiplex a single relay
func (*Consumer) Ack ¶
Ack will send an acknowledgement to the server that the last message returned by Consume was processed. If EnableMultiAck is true, then all messages up to the last consumed one will be acknowledged
func (*Consumer) Close ¶
Close will shutdown the Consumer. Any messages that are still in flight will be Nack'ed.
func (*Consumer) Consume ¶
Consume will consume the next available message. The message must be acknowledged with Ack() or Nack() before the next call to Consume unless EnableMultiAck is true.
func (*Consumer) ConsumeAck ¶
ConsumeAck will consume the next message and acknowledge that the message has been received. This prevents the message from being redelivered, and no call to Ack() or Nack() is needed.
func (*Consumer) ConsumeTimeout ¶
Consume will consume the next available message or times out waiting. The message must be acknowledged with Ack() or Nack() before the next call to Consume unless EnableMultiAck is true.
type GOBSerializer ¶
type GOBSerializer struct {
CustomContentType *string
}
GOBSerializer implements the Serializer interface and uses the GOB format
func (*GOBSerializer) ContentType ¶
func (s *GOBSerializer) ContentType() string
func (*GOBSerializer) RelayDecode ¶
func (*GOBSerializer) RelayDecode(r io.Reader, o interface{}) error
func (*GOBSerializer) RelayEncode ¶
func (*GOBSerializer) RelayEncode(w io.Writer, e interface{}) error
type JSONSerializer ¶
type JSONSerializer struct {
CustomContentType *string
}
JSONSerializer implements the Serializer interface and uses JSON
func (*JSONSerializer) ContentType ¶
func (s *JSONSerializer) ContentType() string
func (*JSONSerializer) RelayDecode ¶
func (*JSONSerializer) RelayDecode(r io.Reader, o interface{}) error
func (*JSONSerializer) RelayEncode ¶
func (*JSONSerializer) RelayEncode(w io.Writer, e interface{}) error
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher is a type that is used only for publishing messages to a single queue. Multiple Publishers can multiplex a single relay.
func (*Publisher) WaitForConfirms ¶
WaitForConfirms blocks until all confirms have arrived or the optional timeout occurs. If the timeout occurs before all confirms have arrived the goroutines blocked on Publish will return immediately with a TimedOut error.
Returns true if all confirms arrived before timeout period. This value is undefined if WaitForConfirms is called from multiple goroutines and Publish is called concurrently.
WaitForConfirms is useful for implementing a graceful exit.
type Relay ¶
func New ¶
New will create a new Relay that can be used to create new publishers or consumers. The caller should no longer modify the configuration once New is invoked, nor should it be shared between multiple relays.
func (*Relay) Close ¶
Close will shutdown the relay. It is best to first Close all the Consumer and Publishers, as this will close the underlying connections.
Source Files
Directories