pika

README

pika

A small RabbitMQ utility wrapper for go services

It provides a simpler abstraction to ease setting up services that communicate with RabbitMQ. To keep simplicity it makes a few assumptions:

• There is only need for Topic Exchanges
• Message body contains JSON
• A Consumer only receives a message type

If you need more features consider a mixed used of the library or simply an alternative.

Connection

To connect to RabbitMQ create a new RabbitConnector

func main() {
  logger, err := zap.NewProduction()
  if err != nil { }
  
  conn := pika.NewConnector(logger)
  if err := conn.Connect(RABBIT_URL); err != nil {
    // TODO
  }
}

This connection will be reused for each Consumer, Publisher and Notifier (1 channel each).

If you have a more complex use-case you can still reuse the connection asking it for a new channel and setting up bindings yourself.

  channel, err := conn.Channel()

Consumer

To receive messages from RabbitMQ create a Consumer of the underlying event type.

type MsgEvent struct {
  Name string `json:"name"`
  Age  int    `json:"age"`
}

type MsgConsumer struct {}

func (c MsgConsumer) Options() pika.ConsumerOptions {
  // TODO
}

func (c MsgConsumer) HandleMessage(e MsgEvent) error {
  return nil
}

Then start the consumer in you main file.

func main() {
  //.. setup logger and connection
  consumer := MsgConsumer{}
  err := pika.StartConsumer[MsgEvent](conn, consumer)
  if err != nil {
    // TODO
  }
  
}

Consumer Options

The connector knows how to setup the connection based on the ConsumerOptions returned by the Consumer. At a minimum it needs an exchange, topic and queue.

func (c MsgConsumer) Options() pika.ConsumerOptions {
  return pika.NewConsumerOptions(
    "", // Exchange
    "", // Topic / Routing-Key
    "", // Queue name (leave empty for random name)
  )
}

To setup the queue to persist if the application crashes.

  return pika.NewConsumerOptions("", "", "").SetDurable()

This allows messages to queue-up and start consuming as soon as it starts

You can also retry messages. It will be done in memory instead of using a dead-letter exchange.

  return pika.NewConsumerOptions("", "", "").WithRetry(1, time.Second)

Publisher

A publisher is a simple abstraction to conform the event type. It only holds a channel.

publisher, err := pika.CreatePublisher[MsgEvent](conn, pika.PublisherOptions{
  Exchange: "",
  Topic: "",
})

// To use
publisher.Publish(MsgEvent{})

Notifier

Sometimes you want to perform operations on regular intervals and them publish the result.

type Notifier[T any] interface {
	Options() NotificationOptions
	Stop() bool
	Notify() (T, error)
}

Documentation

Index

• func AppendToTopic(key, str string) string
• func StartConsumer[T any](r *RabbitConnector, consumer Consumer[T]) error
• func StartNotifier[T any](r *RabbitConnector, notifier Notifier[T]) error
• type Channel
• type Consumer
• type ConsumerOptions
  • func NewConsumerOptions(exchange, topic, queue string) ConsumerOptions
  • func (co ConsumerOptions) HasRetry() bool
  • func (co ConsumerOptions) SetDurable() ConsumerOptions
  • func (co ConsumerOptions) SetName(name string) ConsumerOptions
  • func (co ConsumerOptions) WithRetry(retries int, interval time.Duration) ConsumerOptions
• type Msg
  • func (msg Msg[T]) Retry(backoff time.Duration, msgs chan Msg[T])
  • func (msg Msg[T]) ShouldRetry(retries int) bool
• type NotificationOptions
• type Notifier
• type Publisher
  • func CreatePublisher[T any](r *RabbitConnector, options PublisherOptions) (*Publisher[T], error)
  • func (p Publisher[T]) Publish(message T) error
• type PublisherOptions
• type Queue
  • func NewQueue[T any]() *Queue[T]
  • func (q *Queue[T]) Dequeue() *T
  • func (q *Queue[T]) DequeueNoWait() *T
  • func (q *Queue[T]) Length() int
  • func (q *Queue[T]) PeakAt(i int) *T
  • func (q *Queue[T]) Peek() *T
  • func (q *Queue[T]) Queue(e *T)
• type RabbitConnector
  • func NewConnector(logger *zap.Logger) *RabbitConnector
  • func (rc *RabbitConnector) Channel() (*Channel, error)
  • func (rc *RabbitConnector) Connect(url string) error
  • func (rc RabbitConnector) Disconnect() error

Functions

func AppendToTopic

func AppendToTopic(key, str string) string

func StartConsumer

func StartConsumer[T any](r *RabbitConnector, consumer Consumer[T]) error

StartConsumer makes the necessary declarations and bindings to start a consumer. It will spawn 2 goroutines to receive and process (with retries) messages.

Everything will be handle with the options declared in the `Consumer` method `Options`

func StartNotifier

func StartNotifier[T any](r *RabbitConnector, notifier Notifier[T]) error

StartNotifier initiates the `Notifier` that will run in a goroutine

Types

type Channel

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

Wraps an amqp.Channel to handle reconnects

type Consumer

type Consumer[T any] interface {
	Options() ConsumerOptions
	HandleMessage(T) error
}

Consumer represents a RabbitMQ consumer for a typed `T` message

type ConsumerOptions

type ConsumerOptions struct {
	Exchange  string
	Topic     string
	QueueName string
	// contains filtered or unexported fields
}

ConsumerOptions represents a queue binding for a Consumer

func NewConsumerOptions

func NewConsumerOptions(exchange, topic, queue string) ConsumerOptions

NewConsumerOptions creates a ConsumerOptions object with default configurations

func (ConsumerOptions) HasRetry

func (co ConsumerOptions) HasRetry() bool

func (ConsumerOptions) SetDurable

func (co ConsumerOptions) SetDurable() ConsumerOptions

SetDurable configures the queue to be persist if the consumer disconnects

func (ConsumerOptions) SetName

func (co ConsumerOptions) SetName(name string) ConsumerOptions

SetName sets the consumer name

func (ConsumerOptions) WithRetry

func (co ConsumerOptions) WithRetry(retries int, interval time.Duration) ConsumerOptions

WithRetry enables in memory retries of unhandled messages. It will retry `retries` times waiting `interval` each time.

type Msg

type Msg[T any] struct {
	// contains filtered or unexported fields
}

Msg holds a RabbitMQ Delivery content's (after parse) alonside any attacked state needed. ex: number of retries

func (Msg[T]) Retry

func (msg Msg[T]) Retry(backoff time.Duration, msgs chan Msg[T])

Retry increases inner counter and sleeps before sending itself back through msgs. Use a goroutine!

func (Msg[T]) ShouldRetry

func (msg Msg[T]) ShouldRetry(retries int) bool

ShouldRetry reports if a msg can still be retried

type NotificationOptions

type NotificationOptions struct {
	Exchange string
	Topic    string
	Interval time.Duration
}

NotificationOptions represents the behaviour of the `Notifier`

type Notifier

type Notifier[T any] interface {
	Options() NotificationOptions
	Stop() bool
	Notify() (T, error)
}

Notifier is a `Publisher` that publish on a regular interval

type Publisher

type Publisher[T any] struct {
	// contains filtered or unexported fields
}

Publisher represents a specific msg that can be published

func CreatePublisher

func CreatePublisher[T any](r *RabbitConnector, options PublisherOptions) (*Publisher[T], error)

CreatePublisher creates a `Publisher`

func (Publisher[T]) Publish

func (p Publisher[T]) Publish(message T) error

Publish publishes the `message` on the specified exchange and queue

type PublisherOptions

type PublisherOptions struct {
	Exchange string
	Topic    string
}

PublisherOptions specifies where a Publisher will publish messages

type Queue

type Queue[T any] struct {
	// contains filtered or unexported fields
}

func NewQueue

func NewQueue[T any]() *Queue[T]

func (*Queue[T]) Dequeue

func (q *Queue[T]) Dequeue() *T

func (*Queue[T]) DequeueNoWait

func (q *Queue[T]) DequeueNoWait() *T

func (*Queue[T]) Length

func (q *Queue[T]) Length() int

func (*Queue[T]) PeakAt

func (q *Queue[T]) PeakAt(i int) *T

func (*Queue[T]) Peek

func (q *Queue[T]) Peek() *T

func (*Queue[T]) Queue

func (q *Queue[T]) Queue(e *T)

type RabbitConnector

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

func NewConnector

func NewConnector(logger *zap.Logger) *RabbitConnector

func (*RabbitConnector) Channel

func (rc *RabbitConnector) Channel() (*Channel, error)

func (*RabbitConnector) Connect

func (rc *RabbitConnector) Connect(url string) error

func (RabbitConnector) Disconnect

func (rc RabbitConnector) Disconnect() error