README
¶
Carrot
Example
import (
"fmt"
"github.com/PMoneda/carrot"
)
func main(){
config := carrot.ConnectionConfig{
Host: os.Getenv("RABBITMQ_HOST"),
Username: os.Getenv("RABBITMQ_USERNAME"),
Password: os.Getenv("RABBITMQ_PASSWORD"),
VHost: "/",
}
exchangeName := "my_exchange"
var conn *carrot.BrokerClient
builder = carrot.NewBuilder(conn)
builder.UseVHost("plataforma_v1.0")
builder.DeclareTopicExchange(exchangeName)
builder.DeclareTopicExchange(exchangeName + "_error")
builder.UpdateTopicPermission(config.Username, exchangeName)
builder.UpdateTopicPermission(config.Username, exchangeName+"_error")
subConn, _ := carrot.NewBrokerClient(&config)
subscriber = carrot.NewSubscriber(subConn)
pubConn, _ := carrot.NewBrokerClient(&config)
publisher = carrot.NewPublisher(pubConn)
pickerConn, _ := carrot.NewBrokerClient(&config)
picker = carrot.NewPicker(pickerConn)
}
Documentation
¶
Index ¶
- type BrokerClient
- type Builder
- func (builder *Builder) BindQueueToExchange(queue, exchange, routingKey string) error
- func (builder *Builder) DeclareDirectExchange(exchange string) error
- func (builder *Builder) DeclareFanoutExchange(exchange string) error
- func (builder *Builder) DeclareHeadersExchange(exchange string) error
- func (builder *Builder) DeclareQueue(queue string) error
- func (builder *Builder) DeclareTopicExchange(exchange string) error
- func (builder *Builder) UpdateTopicPermission(user, exchange string) error
- func (builder *Builder) UseVHost(vhost string) error
- type ConnectionConfig
- type Message
- type MessageContext
- type Picker
- type Publisher
- type SubscribeWorker
- type Subscriber
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BrokerClient ¶
type BrokerClient struct {
// contains filtered or unexported fields
}
BrokerClient is a struct to manager api and ampq connection
func NewBrokerClient ¶
func NewBrokerClient(config *ConnectionConfig) (client *BrokerClient, err error)
NewBrokerClient creates a new rabbit broker client
func (*BrokerClient) Channel ¶
func (broker *BrokerClient) Channel() (ch *amqp.Channel, err error)
Channel return amqp channel with reconnect capabilities
func (*BrokerClient) Reconnect ¶
func (broker *BrokerClient) Reconnect() (err error)
Reconnect to rabbitmq
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder builds a rabbit infrastructure
func NewBuilder ¶
func NewBuilder(conn *BrokerClient) *Builder
NewBuilder creates new broker builder
func (*Builder) BindQueueToExchange ¶
BindQueueToExchange binds a queue to a exchange
func (*Builder) DeclareDirectExchange ¶
DeclareDirectExchange create a durable direct exchange
func (*Builder) DeclareFanoutExchange ¶
DeclareFanoutExchange create a durable fanout exchange
func (*Builder) DeclareHeadersExchange ¶
DeclareHeadersExchange create a durable headers exchange
func (*Builder) DeclareQueue ¶
DeclareQueue creates a durable queue
func (*Builder) DeclareTopicExchange ¶
DeclareTopicExchange create a durable topic exchange
func (*Builder) UpdateTopicPermission ¶
UpdateTopicPermission updates or create a new topic permission
type ConnectionConfig ¶
type ConnectionConfig struct {
Username string
Password string
Host string
AMQPPort string
APIPort string
VHost string
}
func (*ConnectionConfig) GetAMQPPort ¶
func (conn *ConnectionConfig) GetAMQPPort() string
GetAMQPPort returns current config of amqp port or default port
func (*ConnectionConfig) GetAMQPURI ¶
func (conn *ConnectionConfig) GetAMQPURI() string
GetAMQPURI returns amqp url format based on config
func (*ConnectionConfig) GetAPIPort ¶
func (conn *ConnectionConfig) GetAPIPort() string
GetAPIPort returns current api port or default
func (*ConnectionConfig) GetAPIURI ¶
func (conn *ConnectionConfig) GetAPIURI() string
GetAPIURI return uri formatted
type Message ¶
type Message struct {
Data []byte
ContentType string
Encoding string
Headers map[string]interface{}
}
Message encapsulate some data configuration
type MessageContext ¶
type MessageContext struct {
Message Message
// contains filtered or unexported fields
}
MessageContext manager received message from rabbit and ack process
func (*MessageContext) Nack ¶
func (ctx *MessageContext) Nack(requeue bool) error
Nack message to server if requeue = true the message will be sent to same queue
func (*MessageContext) RedirectTo ¶
func (ctx *MessageContext) RedirectTo(exchange, routingKey string) error
RedirectTo redirect message to other exchange
type Picker ¶
type Picker struct {
// contains filtered or unexported fields
}
Picker just pick message from queue without continuous consuming
func NewPicker ¶
func NewPicker(client *BrokerClient) *Picker
NewPicker creates a new broker queue picker
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
func NewPublisher ¶
func NewPublisher(client *BrokerClient) *Publisher
NewPublisher creates a new broker publisher
type SubscribeWorker ¶
type SubscribeWorker struct {
Scale uint
Handler func(*MessageContext) error
AutoAck bool
Queue string
}
SubscribeWorker is the worker handler for queues
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber is a consumer component to Rabbit
func NewSubscriber ¶
func NewSubscriber(client *BrokerClient) *Subscriber
NewSubscriber creates a new Subscriber for Rabbit
func (*Subscriber) SetMaxRetries ¶
func (sub *Subscriber) SetMaxRetries(n int)
SetMaxRetries on reconnecting subscriber
func (*Subscriber) Subscribe ¶
func (sub *Subscriber) Subscribe(worker SubscribeWorker) error
Subscribe binds a worker to queue on Rabbit
Source Files
Directories