consumer

Documentation

Index

• Variables
• type Config
  • func NewConfig(address string, queue string, handler Handler, opts ...Option) Config
  • func (c Config) String() string
• type ConnOption
• type Consumer
  • func New(config Config) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Run() error
• type Delivery
  • func NewDelivery(donner Donner, conn *stomp.Conn, source *stomp.Message) *Delivery
  • func (d *Delivery) Ack() error
  • func (d *Delivery) Nack() error
  • func (d *Delivery) Source() *stomp.Message
• type Donner
• type Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, delivery *Delivery)
• type Middleware
• type NoopObserver
  • func (n NoopObserver) BeginConsuming(c *Consumer)
  • func (n NoopObserver) CloseDone(c *Consumer)
  • func (n NoopObserver) CloseStart(c *Consumer)
  • func (n NoopObserver) Error(c *Consumer, err error)
• type Observer
• type Option
  • func WithConcurrency(concurrency int) Option
  • func WithConnectionOptions(connOpts ...ConnOption) Option
  • func WithMiddlewares(middlewares ...Middleware) Option
  • func WithObserver(observer Observer) Option
  • func WithSubscriptionOptions(subOpts ...SubscriptionOption) Option
• type SubscriptionOption
• type Watcher
  • func NewWatcher(config Config) *Watcher
  • func (w *Watcher) Run(ctx context.Context) error
  • func (w *Watcher) Shutdown()

Variables

var (
	ErrDeliveryAlreadyHandled = errors.New("delivery already handled")
)

Types

type Config

type Config struct {
	Address          string
	Queue            string
	ConnOpts         []ConnOption
	Concurrency      int
	Middlewares      []Middleware
	SubscriptionOpts []SubscriptionOption
	Observer         Observer
	ReconnectTimeout time.Duration
	// contains filtered or unexported fields
}

func NewConfig

func NewConfig(address string, queue string, handler Handler, opts ...Option) Config

func (Config) String

func (c Config) String() string

type ConnOption

type ConnOption = func(*stomp.Conn) error

type Consumer

type Consumer struct {
	Config
	// contains filtered or unexported fields
}

func New

func New(config Config) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Run

func (c *Consumer) Run() error

type Delivery

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

func NewDelivery

func NewDelivery(donner Donner, conn *stomp.Conn, source *stomp.Message) *Delivery

func (*Delivery) Ack

func (d *Delivery) Ack() error

func (*Delivery) Nack

func (d *Delivery) Nack() error

func (*Delivery) Source

func (d *Delivery) Source() *stomp.Message

type Donner

type Donner interface {
	Done()
}

type Handler

type Handler interface {
	Handle(ctx context.Context, delivery *Delivery)
}

type HandlerFunc

type HandlerFunc func(ctx context.Context, delivery *Delivery)

func (HandlerFunc) Handle

func (f HandlerFunc) Handle(ctx context.Context, delivery *Delivery)

type Middleware

type Middleware func(next Handler) Handler

type NoopObserver

type NoopObserver struct {
}

func (NoopObserver) BeginConsuming

func (n NoopObserver) BeginConsuming(c *Consumer)

func (NoopObserver) CloseDone

func (n NoopObserver) CloseDone(c *Consumer)

func (NoopObserver) CloseStart

func (n NoopObserver) CloseStart(c *Consumer)

func (NoopObserver) Error

func (n NoopObserver) Error(c *Consumer, err error)

type Observer

type Observer interface {
	Error(c *Consumer, err error)
	CloseStart(c *Consumer)
	CloseDone(c *Consumer)
	BeginConsuming(c *Consumer)
}

type Option

type Option func(c *Config)

func WithConcurrency

func WithConcurrency(concurrency int) Option

func WithConnectionOptions

func WithConnectionOptions(connOpts ...ConnOption) Option

func WithMiddlewares

func WithMiddlewares(middlewares ...Middleware) Option

func WithObserver

func WithObserver(observer Observer) Option

func WithSubscriptionOptions

func WithSubscriptionOptions(subOpts ...SubscriptionOption) Option

type SubscriptionOption

type SubscriptionOption = func(*frame.Frame) error

type Watcher

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

func NewWatcher

func NewWatcher(config Config) *Watcher

func (*Watcher) Run

func (w *Watcher) Run(ctx context.Context) error

func (*Watcher) Shutdown

func (w *Watcher) Shutdown()

Shutdown Perform graceful shutdown