rabbitmq

Documentation

Index

• Constants
• Variables
• func RetryCountFor(e *ziggurat.Event) int
• type ARetry
  • func AutoRetry(qc Queues, opts ...Opts) *ARetry
  • func (r *ARetry) Consume(ctx context.Context, h ziggurat.Handler) error
  • func (r *ARetry) DSReplayHandler(ctx context.Context) http.Handler
  • func (r *ARetry) DSViewHandler(ctx context.Context) http.Handler
  • func (r *ARetry) DeleteQueuesAndExchanges(ctx context.Context, queueName string) error
  • func (r *ARetry) InitPublishers(ctx context.Context) error
  • func (r *ARetry) Publish(ctx context.Context, event *ziggurat.Event, queueKey string, queueType string, ...) error
  • func (r *ARetry) Retry(ctx context.Context, event *ziggurat.Event, queueKey string) error
• type Opts
  • func WithConnectionTimeout(t time.Duration) Opts
  • func WithHosts(hosts ...string) Opts
  • func WithLogger(l ziggurat.StructuredLogger) Opts
  • func WithPassword(password string) Opts
  • func WithUsername(username string) Opts
• type QueueConfig
• type Queues

Constants

const (
	QueueTypeDL      = "dlq"
	QueueTypeInstant = "instant"
	QueueTypeDelay   = "delay"
)

const KeyRetryCount = "rabbitmqAutoRetryCount"

Variables

var ErrCleanShutdown = errors.New("clean shutdown of rabbitmq streams")

var ErrPublisherNotInit = errors.New("auto retry publish error: publisher not initialized, please call the InitPublisher method")

Functions

func RetryCountFor

func RetryCountFor(e *ziggurat.Event) int

Types

type ARetry

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

func AutoRetry

func AutoRetry(qc Queues, opts ...Opts) *ARetry

func (*ARetry) Consume

func (r *ARetry) Consume(ctx context.Context, h ziggurat.Handler) error

func (*ARetry) DSReplayHandler

func (r *ARetry) DSReplayHandler(ctx context.Context) http.Handler

func (*ARetry) DSViewHandler

func (r *ARetry) DSViewHandler(ctx context.Context) http.Handler

DSViewHandler allows you to peek into

the rabbitMQ dead-set queue.

func (*ARetry) DeleteQueuesAndExchanges

func (r *ARetry) DeleteQueuesAndExchanges(ctx context.Context, queueName string) error

func (*ARetry) InitPublishers

func (r *ARetry) InitPublishers(ctx context.Context) error

func (*ARetry) Publish

func (r *ARetry) Publish(ctx context.Context, event *ziggurat.Event, queueKey string, queueType string, expirationMS string) error

Publish can be called from anywhere and messages can be sent to any queue

func (*ARetry) Retry

func (r *ARetry) Retry(ctx context.Context, event *ziggurat.Event, queueKey string) error

type Opts

type Opts func(r *ARetry)

func WithConnectionTimeout

func WithConnectionTimeout(t time.Duration) Opts

func WithHosts

func WithHosts(hosts ...string) Opts

func WithLogger

func WithLogger(l ziggurat.StructuredLogger) Opts

func WithPassword

func WithPassword(password string) Opts

func WithUsername

func WithUsername(username string) Opts

type QueueConfig

type QueueConfig struct {
	QueueKey              string
	DelayExpirationInMS   string
	RetryCount            int
	ConsumerPrefetchCount int
	ConsumerCount         int
}

type Queues

type Queues []QueueConfig