amqpsubscriber

Documentation

Index

• Constants
• func New(connOpt ConnectionOption, opts ...Option) easypubsub.Subscriber
• type ConnectionOption
  • func Connection(url string) ConnectionOption
  • func ConnectionWithConfig(url string, amqpConfig *amqp.Config) ConnectionOption
  • func ConnectionWithTLS(url string, tlsConfig *tls.Config) ConnectionOption
• type Consume
• type Exchange
• type Option
  • func WithConsume(consume *Consume) Option
  • func WithExchange(exchange *Exchange) Option
  • func WithLogger(logger easypubsub.Logger) Option
  • func WithQosConfig(qosConfig *QosConfig) Option
  • func WithQueue(queue *Queue) Option
  • func WithQueueBinds(queueBinds ...*QueueBind) Option
  • func WithReconnectBackoff(reconnectBackoff backoffutils.BackoffFunc) Option
  • func WithRequeueOnNack(enable bool) Option
  • func WithUnmarshalMsgFunc(unmarshalMsgFunc UnmarshalMsgFunc) Option
• type QosConfig
• type Queue
• type QueueBind
• type Subscriber
  • func (sub *Subscriber) Close() error
  • func (sub *Subscriber) String() string
  • func (sub *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *easypubsub.Message, <-chan error)
• type UnmarshalMsgFunc

Constants

const (
	NORMAL = 0
	CLOSED = 1
)

Functions

func New

func New(connOpt ConnectionOption, opts ...Option) easypubsub.Subscriber

Types

type ConnectionOption

type ConnectionOption func(o *connectionOptions)

func Connection

func Connection(url string) ConnectionOption

func ConnectionWithConfig

func ConnectionWithConfig(url string, amqpConfig *amqp.Config) ConnectionOption

func ConnectionWithTLS

func ConnectionWithTLS(url string, tlsConfig *tls.Config) ConnectionOption

type Consume

type Consume struct {
	Consumer  string
	Exclusive bool
	NoLocal   bool
	NoWait    bool
	Args      map[string]interface{}
}

type Exchange

type Exchange struct {
	NameFunc   func(topic string) string
	Kind       string
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Args       map[string]interface{}
}

type Option

type Option func(o *options)

func WithConsume

func WithConsume(consume *Consume) Option

func WithExchange

func WithExchange(exchange *Exchange) Option

func WithLogger

func WithLogger(logger easypubsub.Logger) Option

func WithQosConfig

func WithQosConfig(qosConfig *QosConfig) Option

func WithQueue

func WithQueue(queue *Queue) Option

func WithQueueBinds

func WithQueueBinds(queueBinds ...*QueueBind) Option

func WithReconnectBackoff

func WithReconnectBackoff(reconnectBackoff backoffutils.BackoffFunc) Option

func WithRequeueOnNack

func WithRequeueOnNack(enable bool) Option

func WithUnmarshalMsgFunc

func WithUnmarshalMsgFunc(unmarshalMsgFunc UnmarshalMsgFunc) Option

type QosConfig

type QosConfig struct {
	PrefetchCount int
	PrefetchSize  int
	Global        bool
}

type Queue

type Queue struct {
	NameFunc   func(topic string) string
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Args       map[string]interface{}
}

type QueueBind

type QueueBind struct {
	Key    string
	NoWait bool
	Args   map[string]interface{}
}

type Subscriber

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

func (*Subscriber) Close

func (sub *Subscriber) Close() error

func (*Subscriber) String

func (sub *Subscriber) String() string

func (*Subscriber) Subscribe

func (sub *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *easypubsub.Message, <-chan error)

type UnmarshalMsgFunc

type UnmarshalMsgFunc func(ctx context.Context, topic string, amqpMsg *amqp.Delivery) (msg *easypubsub.Message, err error)