rabbitmqsmartshovel

Documentation

Index

• Variables
• type Config
• type Session
  • func New(logger log.Logger, config Config) *Session
  • func (session *Session) Ack(tag uint64) error
  • func (session *Session) Close() error
  • func (session *Session) Nack(tag uint64) error
  • func (session *Session) Push(routingKey string, data amqp.Publishing) error
  • func (session *Session) Stream(consumergroup string, autoAck bool, exclusive bool) (chan amqp.Delivery, error)

Variables

var (
	// ErrNotConnected is returned when not connected to a server
	ErrNotConnected = errors.New("not connected to a server")
	// ErrAlreadyClosed is returned when the connection is already closed
	ErrAlreadyClosed = errors.New("already closed: not connected to the server")
	// ErrShutdown is returned when already shutting down
	ErrShutdown = errors.New("session is shutting down")
	// ErrNack is returned when a message publish fails with a NACK
	ErrNack = errors.New("message was not sent")
	// ErrServerBusy is returned when there is too much tcp backpressure on a channel
	ErrServerBusy = errors.New("server busy; message was not sent")
	// ErrPublishOnly is returned when a channel is publish only and you try and use a queue function
	ErrPublishOnly = errors.New("channel is publish only")
	// ErrTimedOut is returned when a message times out waiting for confirmation
	ErrTimedOut = errors.New("confirmation timed out")
)

Types

type Config

type Config struct {
	Name                        string
	ID                          string
	Exchange                    string
	IngestExchange              string
	SubscriptionExchange        string
	IsShovel                    bool
	IsSubscriptionQueueListener bool
	ConsumerConnectionPool      *rabbitmq.ConnectionPool
	PublisherConnectionPool     *rabbitmq.ConnectionPool
	AutoAck                     bool
	DurableQueue                bool
	DeleteUnused                bool
	Exclusive                   bool
	Args                        amqp.Table
	Qos                         int
	PublishOnly                 bool
	Context                     context.Context
}

Config for the session

type Session

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

Session is the rabbitmq session

func New

func New(logger log.Logger, config Config) *Session

New creates a new consumer state instance, and automatically attempts to connect to the server.

func (*Session) Ack

func (session *Session) Ack(tag uint64) error

Ack a consumer tag

func (*Session) Close

func (session *Session) Close() error

Close will cleanly shutdown the channel and connection.

func (*Session) Nack

func (session *Session) Nack(tag uint64) error

Nack a consumer tag

func (*Session) Push

func (session *Session) Push(routingKey string, data amqp.Publishing) error

Push will publish data to channel

func (*Session) Stream

func (session *Session) Stream(consumergroup string, autoAck bool, exclusive bool) (chan amqp.Delivery, error)

Stream will continuously put queue items on the channel. It is required to call delivery.Ack when it has been successfully processed, or delivery.Nack when it fails. Ignoring this will cause data to build up on the server.