middleware

Documentation

Index

• Constants
• Variables
• func CorrelationID(h message.HandlerFunc) message.HandlerFunc
• func InstantAck(h message.HandlerFunc) message.HandlerFunc
• func MessageCorrelationID(message *message.Message) string
• func RandomFail(errorRatio float32) message.HandlerMiddleware
• func RandomPanic(panicRatio float32) message.HandlerMiddleware
• func Recoverer(h message.HandlerFunc) message.HandlerFunc
• func SetCorrelationID(id string, msg *message.Message)
• type IgnoreErrors
  • func NewIgnoreErrors(errs []error) IgnoreErrors
  • func (i IgnoreErrors) Middleware(h message.HandlerFunc) message.HandlerFunc
• type Metrics
  • func NewMetrics(timer metrics.Timer, errs metrics.Counter, success metrics.Counter) Metrics
  • func (m Metrics) Middleware(h message.HandlerFunc) message.HandlerFunc
  • func (m Metrics) ShowStats(interval time.Duration, logger metrics.Logger)
• type OnRetryHook
• type PoisonQueue
  • func NewPoisonQueue(pub message.Publisher, topic string) (PoisonQueue, error)
• type Retry
  • func (r Retry) Middleware(h message.HandlerFunc) message.HandlerFunc
• type Throttle
  • func NewThrottlePerSecond(perSecond int, logger watermill.LoggerAdapter) (Throttle, error)
  • func (t Throttle) Middleware(h message.HandlerFunc) message.HandlerFunc

Constants

const CorrelationIDMetadataKey = "correlation_id"

const RetryForever = -1

Variables

var ErrInvalidPoisonQueueTopic = errors.New("invalid poison queue topic")

ErrInvalidPoisonQueueTopic occurs when the topic supplied to the PoisonQueue constructor is invalid.

var ReasonForPoisonedKey = "reason_poisoned"

ReasonForPoisonedKey is the metadata key which marks the reason (error) why the message was deemed poisoned.

Functions

func CorrelationID

func CorrelationID(h message.HandlerFunc) message.HandlerFunc

func InstantAck

func InstantAck(h message.HandlerFunc) message.HandlerFunc

func MessageCorrelationID

func MessageCorrelationID(message *message.Message) string

func RandomFail

func RandomFail(errorRatio float32) message.HandlerMiddleware

func RandomPanic

func RandomPanic(panicRatio float32) message.HandlerMiddleware

func Recoverer

func Recoverer(h message.HandlerFunc) message.HandlerFunc

func SetCorrelationID

func SetCorrelationID(id string, msg *message.Message)

Types

type IgnoreErrors

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

func NewIgnoreErrors

func NewIgnoreErrors(errs []error) IgnoreErrors

func (IgnoreErrors) Middleware

func (i IgnoreErrors) Middleware(h message.HandlerFunc) message.HandlerFunc

type Metrics

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

todo - rewrite (more universal?)

func NewMetrics

func NewMetrics(timer metrics.Timer, errs metrics.Counter, success metrics.Counter) Metrics

func (Metrics) Middleware

func (m Metrics) Middleware(h message.HandlerFunc) message.HandlerFunc

func (Metrics) ShowStats

func (m Metrics) ShowStats(interval time.Duration, logger metrics.Logger)

type OnRetryHook

type OnRetryHook func(retryNum int, delay time.Duration)

type PoisonQueue

type PoisonQueue struct {
	Middleware message.HandlerMiddleware
	// contains filtered or unexported fields
}

PoisonQueue provides a middleware that salvages unprocessable messages and published them on a separate topic. The main middleware chain then continues on, business as usual.

func NewPoisonQueue

func NewPoisonQueue(pub message.Publisher, topic string) (PoisonQueue, error)

type Retry

type Retry struct {
	MaxRetries int

	WaitTime time.Duration
	Backoff  int64

	MaxDelay time.Duration

	OnRetryHook OnRetryHook

	Logger watermill.LoggerAdapter
}

func (Retry) Middleware

func (r Retry) Middleware(h message.HandlerFunc) message.HandlerFunc

type Throttle

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

func NewThrottlePerSecond

func NewThrottlePerSecond(perSecond int, logger watermill.LoggerAdapter) (Throttle, error)

func (Throttle) Middleware

func (t Throttle) Middleware(h message.HandlerFunc) message.HandlerFunc