pika

Documentation

Index

• func AppendToTopic(key, str string) string
• func StartConsumer[T any](consumer Consumer[T], r *RabbitConnector) 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) WithRetry(retries int, interval time.Duration) ConsumerOptions
• type ConsumerRetry
• 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() *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](consumer Consumer[T], r *RabbitConnector) error

func StartNotifier

func StartNotifier[T any](r *RabbitConnector, notifier Notifier[T]) 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[T any] interface {
	Options() NotificationOptions
	Stop() bool
	Notify() (T, error)
}

type Publisher

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

func CreatePublisher

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

func (Publisher[T]) Publish

func (p Publisher[T]) Publish(message T) 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) Disconnect

func (rc RabbitConnector) Disconnect() error