pika

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 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() (*amqp.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 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() (*amqp.Channel, error)

func (*RabbitConnector) Connect

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

func (RabbitConnector) Disconnect

func (rc RabbitConnector) Disconnect() error