pika

Documentation

Index

• func AppendToTopic(key, str string) string
• func StartConsumer[T any](consumer Consumer[T], r *RabbitConnector) error
• type Consumer
• type ConsumerOptions
  • func NewConsumerOptions(exchange, topic, queue string) ConsumerOptions
  • func (co ConsumerOptions) HasRetry() bool
  • func (co ConsumerOptions) WithRetry(retries int, interval time.Duration) ConsumerOptions
• type ConsumerRetry
• 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]) 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() *RabbitConnector
  • 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) StartNotifier(notifier Notifier) error

Functions

func AppendToTopic

func AppendToTopic(key, str string) string

func StartConsumer

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

Types

type Consumer

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

type ConsumerOptions

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

func NewConsumerOptions

func NewConsumerOptions(exchange, topic, queue string) ConsumerOptions

func (ConsumerOptions) HasRetry

func (co ConsumerOptions) HasRetry() bool

func (ConsumerOptions) WithRetry

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

type ConsumerRetry

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

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]) 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() *RabbitConnector

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) StartNotifier

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