astiamqp

README

Astiamqp

Wrapper on top of amqp to provide proper configuration and error handling

Usage

// Setup logger
l := log.New(log.Writer(), log.Prefix(), log.Flags())

// Create worker
w := astikit.NewWorker(astikit.WorkerOptions{Logger: l})

// Create amqp
a := astiamqp.New(c, l)
defer a.Close()

// You can create producers before and after starting amqp
p, _ := a.NewProducer(astiamqp.ConfigurationProducer{
    Exchange: astiamqp.ConfigurationExchange{
        Durable: true,
        Name:    "my-exchange",
        Type:    astiamqp.ExchangeTypeTopic,
    },
})

// You can add consumers before and after starting amqp
a.AddConsumer(astiamqp.ConfigurationConsumer{
    AutoAck: false,
    Exchange: astiamqp.ConfigurationExchange{
        Durable: true,
        Name:    "my-exchange",
        Type:    astiamqp.ExchangeTypeTopic,
    },
    Handler: myHandler,
    Queue: astiamqp.ConfigurationQueue{
        Durable: true,
        Name:    "my-queue",
    },
    RoutingKey: "my.routing.key",
})

// You can publish messages before and after starting amqp.
// If you publish a message before starting amqp, it will be buffered and will be sent once 
// it's connected to the channel
p.Publish("my payload", "my.routing.key")

// Start amqp
a.Start(w)

// Stop amqp
a.Stop()

Documentation

Index

• Variables
• type AMQP
  • func New(c Configuration, l astikit.StdLogger) (a *AMQP)
  • func (a *AMQP) AddConsumer(c ConfigurationConsumer) (err error)
  • func (a *AMQP) Close() error
  • func (a *AMQP) NewProducer(c ConfigurationProducer) (p *Producer, err error)
  • func (a *AMQP) Start(w *astikit.Worker)
  • func (a *AMQP) Stop()
• type Acknowledger
• type Configuration
  • func FlagConfig() Configuration
• type ConfigurationConsumer
• type ConfigurationExchange
• type ConfigurationProducer
• type ConfigurationQOS
• type ConfigurationQueue
• type Consumer
• type ExchangeType
• type Handler
• type Producer
  • func (p *Producer) Produce(msg interface{}, routingKey string) (err error)
• type Table

Variables

var (
	Addr     = flag.String("amqp-addr", "", "the amqp addr")
	Password = flag.String("amqp-password", "", "the amqp password")
	Username = flag.String("amqp-username", "", "the amqp username")
)

Flags

Types

type AMQP

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

AMQP represents a client capable of sending/listening to AMQP queues

func New

func New(c Configuration, l astikit.StdLogger) (a *AMQP)

New creates a new AMQP instance based on a configuration

func (*AMQP) AddConsumer

func (a *AMQP) AddConsumer(c ConfigurationConsumer) (err error)

AddConsumer adds a consumer

func (*AMQP) Close

func (a *AMQP) Close() error

Close closes amqp properly

func (*AMQP) NewProducer

func (a *AMQP) NewProducer(c ConfigurationProducer) (p *Producer, err error)

NewProducer creates a producer

func (*AMQP) Start

func (a *AMQP) Start(w *astikit.Worker)

Start starts amqp

func (*AMQP) Stop

func (a *AMQP) Stop()

Stop stops amqp

type Acknowledger

type Acknowledger interface {
	Ack(multiple bool) error
	Nack(multiple bool, requeue bool) error
	Reject(requeue bool) error
}

Acknowledger represents an acknowledger

type Configuration

type Configuration struct {
	Addr     string            `toml:"addr"`
	Password string            `toml:"password"`
	QOS      *ConfigurationQOS `toml:"qos"`
	Username string            `toml:"username"`
}

Configuration represents the AMQP configuration

func FlagConfig

func FlagConfig() Configuration

FlagConfig returns an AMQP config based on flags

type ConfigurationConsumer

type ConfigurationConsumer struct {
	Arguments  Table
	AutoAck    bool
	Exchange   ConfigurationExchange
	Exclusive  bool
	Handler    Handler
	NoLocal    bool
	NoWait     bool
	Queue      ConfigurationQueue
	RoutingKey string
}

ConfigurationConsumer represents a consumer configuration

type ConfigurationExchange

type ConfigurationExchange struct {
	Arguments   Table
	AutoDeleted bool
	Durable     bool
	Internal    bool
	Name        string
	NoWait      bool
	Type        ExchangeType
}

ConfigurationExchange represents an exchange configuration

type ConfigurationProducer

type ConfigurationProducer struct {
	Exchange ConfigurationExchange
}

ConfigurationProducer represents a producer configuration

type ConfigurationQOS

type ConfigurationQOS struct {
	Global        bool `toml:"global"`
	PrefetchCount int  `toml:"prefetch_count"`
	PrefetchSize  int  `toml:"prefetch_size"`
}

ConfigurationQOS represents the AMQP QOS configuration

type ConfigurationQueue

type ConfigurationQueue struct {
	Arguments   Table
	AutoDeleted bool
	Durable     bool
	Exclusive   bool
	Name        string
	NoWait      bool
}

ConfigurationQueue represents a queue configuration

type Consumer

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

Consumer represents a Consumer

type ExchangeType

type ExchangeType string

ExchangeType represents an exchange type

const (
	ExchangeTypeDirect  ExchangeType = "direct"
	ExchangeTypeFanout  ExchangeType = "fanout"
	ExchangeTypeHeaders ExchangeType = "headers"
	ExchangeTypeTopic   ExchangeType = "topic"
)

Constants

type Handler

type Handler func(msg []byte, routingKey string, a Acknowledger) (context.Context, error)

Handler handles a message

type Producer

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

Producer represents a producer

func (*Producer) Produce

func (p *Producer) Produce(msg interface{}, routingKey string) (err error)

Produce produces a message on a routing key after json.Marshaling it

type Table

type Table amqp.Table

Table wraps amqp.Table