rabbitmq

Documentation

Index

• type Message
  • func NewMessage(delivery amqp.Delivery, codec encoding.Codec) *Message
  • func (m *Message) Ack() (err error)
  • func (m *Message) Body() []byte
  • func (m *Message) Name() string
  • func (m *Message) Reject() (err error)
  • func (m *Message) Status() queue.MessageStatus
  • func (m *Message) Unmarshal(value interface{}) error
• type Queue
  • func NewQueue(name string, opt *QueueOption) (*Queue, error)
  • func (q *Queue) Consumer(opt *queue.ConsumerOption) (*queue.Consumer, error)
  • func (q *Queue) Later(delay time.Duration, messages ...interface{}) (err error)
  • func (q *Queue) Name() string
  • func (q *Queue) Publish(messages ...interface{}) (err error)
  • func (q *Queue) Purge() error
  • func (q *Queue) Size() int
• type QueueOption
• type Worker
  • func NewWorker(id string, q *Queue, opt *queue.ConsumerOption) *Worker
  • func (w *Worker) Daemon(ctx context.Context, handler queue.HandlerFunc) error
  • func (w *Worker) Name() string

Types

type Message

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

func NewMessage

func NewMessage(delivery amqp.Delivery, codec encoding.Codec) *Message

func (*Message) Ack

func (m *Message) Ack() (err error)

func (*Message) Body

func (m *Message) Body() []byte

func (*Message) Name

func (m *Message) Name() string

func (*Message) Reject

func (m *Message) Reject() (err error)

func (*Message) Status

func (m *Message) Status() queue.MessageStatus

func (*Message) Unmarshal

func (m *Message) Unmarshal(value interface{}) error

type Queue

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

func NewQueue

func NewQueue(name string, opt *QueueOption) (*Queue, error)

func (*Queue) Consumer

func (q *Queue) Consumer(opt *queue.ConsumerOption) (*queue.Consumer, error)

func (*Queue) Later

func (q *Queue) Later(delay time.Duration, messages ...interface{}) (err error)

func (*Queue) Name

func (q *Queue) Name() string

func (*Queue) Publish

func (q *Queue) Publish(messages ...interface{}) (err error)

func (*Queue) Purge

func (q *Queue) Purge() error

func (*Queue) Size

func (q *Queue) Size() int

type QueueOption

type QueueOption struct {
	// Connection the connection is used to receive and send messages
	Connection *amqp.Connection
	// URL is a string in the AMQP URI format,
	// if you do not provide the Connection parameter,
	// we will try to create a new connection from the URL
	URL string
	// Codec is using for marshal and unmarshal messages
	// default is gob codec with s2 compression
	Codec encoding.Codec
}

type Worker

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

func NewWorker

func NewWorker(id string, q *Queue, opt *queue.ConsumerOption) *Worker

func (*Worker) Daemon

func (w *Worker) Daemon(ctx context.Context, handler queue.HandlerFunc) error

func (*Worker) Name

func (w *Worker) Name() string