stompx

Documentation

Index

• Constants
• func ConsumerLog(logger log.Logger) consumer.Middleware
• func ConsumerRequestId() consumer.Middleware
• func DefaultConsumer(cfg ConsumerConfig, handler consumer.Handler, logger log.Logger, ...) consumer.Config
• func DefaultPublisher(cfg PublisherConfig, restMiddlewares ...publisher.Middleware) *publisher.Publisher
• func PublisherLog(logger log.Logger) publisher.Middleware
• func PublisherPersistent() publisher.Middleware
• func PublisherRequestId() publisher.Middleware
• type AdapterFunc
  • func (a AdapterFunc) Handle(ctx context.Context, msg *stomp.Message) Result
• type ConsumerConfig
• type ConsumerGroup
  • func NewConsumerGroup() *ConsumerGroup
  • func (g *ConsumerGroup) Close() error
  • func (g *ConsumerGroup) Upgrade(ctx context.Context, consumers ...consumer.Config) error
• type HandlerAdapter
• type LogObserver
  • func NewLogObserver(logger log.Logger) LogObserver
  • func (l LogObserver) BeginConsuming(c *consumer.Consumer)
  • func (l LogObserver) CloseDone(c *consumer.Consumer)
  • func (l LogObserver) CloseStart(c *consumer.Consumer)
  • func (l LogObserver) Error(c *consumer.Consumer, err error)
• type PublisherConfig
• type Result
  • func Ack() Result
  • func Requeue(err error) Result
• type ResultHandler
  • func NewResultHandler(logger log.Logger, adapter HandlerAdapter) ResultHandler
  • func (r ResultHandler) Handle(ctx context.Context, delivery *consumer.Delivery)

Constants

const (
	RequestIdHeader = "x-request-id"
)

Functions

func ConsumerLog

func ConsumerLog(logger log.Logger) consumer.Middleware

func ConsumerRequestId

func ConsumerRequestId() consumer.Middleware

func DefaultConsumer

func DefaultConsumer(cfg ConsumerConfig, handler consumer.Handler, logger log.Logger, restMiddlewares ...consumer.Middleware) consumer.Config

func DefaultPublisher

func DefaultPublisher(cfg PublisherConfig, restMiddlewares ...publisher.Middleware) *publisher.Publisher

func PublisherLog

func PublisherLog(logger log.Logger) publisher.Middleware

func PublisherPersistent

func PublisherPersistent() publisher.Middleware

func PublisherRequestId

func PublisherRequestId() publisher.Middleware

Types

type AdapterFunc

type AdapterFunc func(ctx context.Context, msg *stomp.Message) Result

func (AdapterFunc) Handle

func (a AdapterFunc) Handle(ctx context.Context, msg *stomp.Message) Result

type ConsumerConfig

type ConsumerConfig struct {
	Address       string            `validate:"required" schema:"Адрес брокера"`
	Queue         string            `validate:"required" schema:"Очередь"`
	Concurrency   int               `schema:"Кол-во обработчиков,по умолчанию 1"`
	PrefetchCount int               `schema:"Кол-во предзагруженных сообщений,по умолчанию не используется"`
	Username      string            `schema:"Имя пользователя"`
	Password      string            `schema:"Пароль"`
	ConnHeaders   map[string]string `schema:"Дополнительные параметры подключения"`
}

type ConsumerGroup

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

func NewConsumerGroup

func NewConsumerGroup() *ConsumerGroup

func (*ConsumerGroup) Close

func (g *ConsumerGroup) Close() error

func (*ConsumerGroup) Upgrade

func (g *ConsumerGroup) Upgrade(ctx context.Context, consumers ...consumer.Config) error

type HandlerAdapter

type HandlerAdapter interface {
	Handle(ctx context.Context, msg *stomp.Message) Result
}

type LogObserver

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

func NewLogObserver

func NewLogObserver(logger log.Logger) LogObserver

func (LogObserver) BeginConsuming

func (l LogObserver) BeginConsuming(c *consumer.Consumer)

func (LogObserver) CloseDone

func (l LogObserver) CloseDone(c *consumer.Consumer)

func (LogObserver) CloseStart

func (l LogObserver) CloseStart(c *consumer.Consumer)

func (LogObserver) Error

func (l LogObserver) Error(c *consumer.Consumer, err error)

type PublisherConfig

type PublisherConfig struct {
	Address     string            `validate:"required" schema:"Адрес брокера"`
	Queue       string            `validate:"required" schema:"Очередь"`
	Username    string            `schema:"Имя пользователя"`
	Password    string            `schema:"Пароль"`
	ConnHeaders map[string]string `schema:"Дополнительные параметры подключения"`
}

type Result

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

func Ack

func Ack() Result

func Requeue

func Requeue(err error) Result

type ResultHandler

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

func NewResultHandler

func NewResultHandler(logger log.Logger, adapter HandlerAdapter) ResultHandler

func (ResultHandler) Handle

func (r ResultHandler) Handle(ctx context.Context, delivery *consumer.Delivery)