pika

Documentation

Index

• func AppendToTopic(key, str string) string
• type Consumer
• type ConsumerOptions
• type NotificationOptions
• type Notifier
• type Publisher
  • func (p Publisher) Publish(message string) error
• type PublisherOptions
• type Queue
  • func NewQueue[T any]() *Queue[T]
  • func (q *Queue[T]) Dequeue() *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(url string) (*RabbitConnector, error)
  • func (rc RabbitConnector) Channel() (*amqp.Channel, error)
  • func (rc *RabbitConnector) Connect(url string) error
  • func (r *RabbitConnector) CreatePublisher(options PublisherOptions) (*Publisher, error)
  • func (rc RabbitConnector) Disconnect() error
  • func (r *RabbitConnector) StartConsumer(consumer Consumer) error
  • func (r *RabbitConnector) StartNotifier(notifier Notifier) error
• type Retrier
  • func NewRetrier(c *amqp.Channel) *Retrier
  • func (r *Retrier) Retry(rm *RetryMessage)
• type RetryMessage

Functions

func AppendToTopic

func AppendToTopic(key, str string) string

Types

type Consumer

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

type ConsumerOptions

type ConsumerOptions struct {
	Exchange  string
	Topic     string
	QueueName string

	Retries       int // 0 -> will not queue for retry
	RetryInterval time.Duration
}

type NotificationOptions

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

type Notifier

type Notifier interface {
	Options() NotificationOptions
	Stop() bool
	Notify() (string, error)
}

type Publisher

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

func (Publisher) Publish

func (p Publisher) Publish(message string) error

type PublisherOptions

type PublisherOptions struct {
	Exchange string
	Topic    string
}

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]) 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(url string) (*RabbitConnector, error)

func (RabbitConnector) Channel

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

func (*RabbitConnector) Connect

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

func (*RabbitConnector) CreatePublisher

func (r *RabbitConnector) CreatePublisher(options PublisherOptions) (*Publisher, error)

func (RabbitConnector) Disconnect

func (rc RabbitConnector) Disconnect() error

func (*RabbitConnector) StartConsumer

func (r *RabbitConnector) StartConsumer(consumer Consumer) error

func (*RabbitConnector) StartNotifier

func (r *RabbitConnector) StartNotifier(notifier Notifier) error

type Retrier

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

func NewRetrier

func NewRetrier(c *amqp.Channel) *Retrier

func (*Retrier) Retry

func (r *Retrier) Retry(rm *RetryMessage)

type RetryMessage

type RetryMessage struct {
	Exchange string
	Topic    string
	Message  string

	Retries  int
	Interval time.Duration
	// contains filtered or unexported fields
}