hedwig

README

Note:

Since the release of cloudevents, there is no need to maintain hedwig anymore. Please use the Go SDK instead of using hedwig.

Hedwig Go

Golang emitter and consumer for Hedwig

About Hedwig

Hedwig is an attempt to standardize messaging and events between micro-services.

It uses RabbitMQ as the transport and simplifies the API down significantly.

Documentation

Index

• Constants
• Variables
• type Callback
• type ConsumerSetting
• type Hedwig
  • func New(settings *Settings) *Hedwig
  • func (h *Hedwig) AddQueue(qSetting *QueueSetting, qName string) error
  • func (h *Hedwig) Consume() error
  • func (h *Hedwig) Disconnect() error
  • func (h *Hedwig) Publish(key string, body []byte) (err error)
  • func (h *Hedwig) PublishWithDelay(key string, body []byte, delay time.Duration) (err error)
  • func (h *Hedwig) PublishWithHeaders(key string, body []byte, headers map[string]interface{}) (err error)
• type QueueSetting
  • func DefaultQueueSetting(callback Callback, bindings ...string) *QueueSetting
• type QueueType
• type Settings
  • func DefaultSettings() *Settings

Constants

const (
	PublishChannel   = "publish"
	SubscribeChannel = "subscribe"
)

const (
	MinDeliveryLimit uint = 1
	QosPrefetchCount      = "RMQ_PREFETCH_COUNT"
)

const (
	// For exchanges which use RMQ delay plugin, following type should be used as ExchangeType
	ExchangeTypeDelayed = "x-delayed-message"
	// Delayed Type Exchanges also need an extra arg with following key with the value set to actual ExchangeType
	// like `direct`, `topic` etc
	DelayedExchangeArgKey = "x-delayed-type"
	// Following header should be used when you are publishing to an Delay exchange. The header value will be
	// delay in milliseconds
	DelayHeader = "x-delay"
)

Variables

var (
	ErrNilHedwig          = errors.New("hedwig is nil, use `New` to create an instance")
	ErrAlreadyInitialized = errors.New("already initialized, disconnect before attempting again")
	ErrNoBindings         = errors.New("no bindings provided, need at least one")
	ErrNoConsumerSetting  = errors.New("consumer setting is nil")
)

Types

type Callback

type Callback func(<-chan amqp.Delivery, *sync.WaitGroup)

type ConsumerSetting

type ConsumerSetting struct {
	Queues map[string]*QueueSetting
	// contains filtered or unexported fields
}

type Hedwig

type Hedwig struct {
	sync.Mutex

	Settings *Settings
	Error    error
	// contains filtered or unexported fields
}

func New

func New(settings *Settings) *Hedwig

func (*Hedwig) AddQueue

func (h *Hedwig) AddQueue(qSetting *QueueSetting, qName string) error

func (*Hedwig) Consume

func (h *Hedwig) Consume() error

func (*Hedwig) Disconnect

func (h *Hedwig) Disconnect() error

func (*Hedwig) Publish

func (h *Hedwig) Publish(key string, body []byte) (err error)

func (*Hedwig) PublishWithDelay

func (h *Hedwig) PublishWithDelay(key string, body []byte, delay time.Duration) (err error)

func (*Hedwig) PublishWithHeaders

func (h *Hedwig) PublishWithHeaders(key string, body []byte, headers map[string]interface{}) (err error)

type QueueSetting

type QueueSetting struct {
	Bindings   []string
	Callback   Callback
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoAck      bool
	QueueType  QueueType
	// DeliveryLimit specifies the maximum number of times a message can be redelivered.
	// It is useful for preventing message poisoning by limiting how many times a message
	// can be retried before it is considered "poisoned" and either dead-lettered or discarded.
	// Note: This setting has no effect on Classic queues, which do not support delivery limits.
	// Caution: Once set, the DeliveryLimit cannot be updated during the lifetime of the queue.
	DeliveryLimit uint
}

func DefaultQueueSetting

func DefaultQueueSetting(callback Callback, bindings ...string) *QueueSetting

type QueueType

type QueueType uint

const (
	QueueType_Classic QueueType = 0
	QueueType_Quorum  QueueType = 1
)

type Settings

type Settings struct {
	Exchange          string
	ExchangeType      string
	ExchangeArgs      amqp.Table
	HeartBeatInterval time.Duration
	SocketTimeout     time.Duration
	Host              string
	Port              int
	Vhost             string
	Username          string
	Password          string
	Consumer          *ConsumerSetting
}

func DefaultSettings

func DefaultSettings() *Settings