pubsub

README

DRAFT

Documentation

Index

• Variables
• type Bind
• type Consumer
• type EventHandler
• type Exchange
• type Option
  • func WithDeclare(ex Exchange, q Queue, b Bind) Option
• type Queue
• type Session
  • func New(addr string, opts ...Option) *Session
  • func (session *Session) ChannelClose() error
  • func (session *Session) Close() error
  • func (session *Session) Push(data []byte) error
  • func (session *Session) QueueDeclare(q Queue) *Session
  • func (session *Session) Stream(c *Consumer) (<-chan amqp.Delivery, error)
  • func (session *Session) Subscribe(handler func([]byte) error) error
  • func (session *Session) UnsafePush(data []byte) error

Variables

var (
	ErrNotConnected  = errors.New("not connected to a server")
	ErrAlreadyClosed = errors.New("already closed: not connected to the server")
	ErrShutdown      = errors.New("session is shutting down")
)

Types

type Bind

type Bind struct {
	QueueName    string
	ExchangeName string
	Key          string
}

type Consumer

type Consumer struct {
	Name      string
	QueueName string
	AutoAck   bool
	Exclusive bool

	Handler EventHandler
	// contains filtered or unexported fields
}

type EventHandler

type EventHandler func([]byte) error

type Exchange

type Exchange struct {
	Name       string
	Kind       string
	Durable    bool
	AutoDelete bool
	Internal   bool
}

type Option

type Option func(*Session)

func WithDeclare

func WithDeclare(ex Exchange, q Queue, b Bind) Option

type Queue

type Queue struct {
	Name       string
	Durable    bool
	AutoDelete bool
	Exclusive  bool
}

type Session

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

func New

func New(addr string, opts ...Option) *Session

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

func (*Session) ChannelClose

func (session *Session) ChannelClose() error

func (*Session) Close

func (session *Session) Close() error

Close will cleanly shutdown the channel and connection.

func (*Session) Push

func (session *Session) Push(data []byte) error

Push will push data onto the queue, and wait for a confirm. If no confirms are received until within the resendTimeout, it continuously re-sends messages until a confirm is received. This will block until the server sends a confirm. Errors are only returned if the push action itself fails, see UnsafePush.

func (*Session) QueueDeclare

func (session *Session) QueueDeclare(q Queue) *Session

func (*Session) Stream

func (session *Session) Stream(c *Consumer) (<-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.

func (*Session) Subscribe

func (session *Session) Subscribe(handler func([]byte) error) error

func (*Session) UnsafePush

func (session *Session) UnsafePush(data []byte) error

UnsafePush will push to the queue without checking for confirmation. It returns an error if it fails to connect. No guarantees are provided for whether the server will recieve the message.