rabbitmq

Documentation

Index

• Constants
• Variables
• type Connection
  • func NewConnection(options *ConnectionOptions) Connection
• type ConnectionOptions
• type ExchangeOptions
• type ExchangeType
  • func (t ExchangeType) String() string
  • func (t ExchangeType) Value() ExchangeType
• type Message
  • func NewMessage() *Message
  • func (m *Message) Ack() error
  • func (m *Message) GetAppID() string
  • func (m *Message) GetBody() []byte
  • func (m *Message) GetContentEncoding() string
  • func (m *Message) GetContentType() string
  • func (m *Message) GetCorrelationID() uuid.UUID
  • func (m *Message) GetDeliveryMode() uint8
  • func (m *Message) GetExpiration() time.Duration
  • func (m *Message) GetExpirationString() string
  • func (m *Message) GetHeader(key string) interface{}
  • func (m *Message) GetHeaders() map[string]interface{}
  • func (m *Message) GetID() uuid.UUID
  • func (m *Message) GetPriority() uint8
  • func (m *Message) GetReplyTo() string
  • func (m *Message) GetTimestamp() time.Time
  • func (m *Message) GetType() string
  • func (m *Message) GetUserID() string
  • func (m *Message) Nack() error
  • func (m *Message) Reject() error
  • func (m *Message) SetAppID(appID string) *Message
  • func (m *Message) SetBody(body []byte) *Message
  • func (m *Message) SetContentEncoding(v string) *Message
  • func (m *Message) SetContentType(v string) *Message
  • func (m *Message) SetCorrelationID(id uuid.UUID) *Message
  • func (m *Message) SetDeliveryModePersistent() *Message
  • func (m *Message) SetExpiration(expiration time.Duration) *Message
  • func (m *Message) SetHeader(key string, value interface{}) *Message
  • func (m *Message) SetPriority(priority uint8) *Message
  • func (m *Message) SetReplyTo(v string) *Message
  • func (m *Message) SetTimestamp(timestamp time.Time) *Message
  • func (m *Message) SetType(v string) *Message
  • func (m *Message) SetUserID(useID string) *Message
• type PublishOptions
• type Publisher
  • func NewPublisher(options *PublisherOptions) Publisher
• type PublisherOptions
• type QueueOptions
  • func (o *QueueOptions) GetArgs() amqp.Table
• type Subscriber
  • func NewSubscriber(options *SubscriberOptions) Subscriber
• type SubscriberOptions

Constants

const ReconnectInterval = 1 * time.Second

Variables

var (
	ErrMessageIsNotDelivery = errors.New("message is not delivery")
)

Types

type Connection

type Connection interface {
	GetConn() *amqp.Connection
	GetChannel() *amqp.Channel
	SetReconnectHooks(...func())
}

func NewConnection

func NewConnection(options *ConnectionOptions) Connection

type ConnectionOptions

type ConnectionOptions struct {
	URI string
}

type ExchangeOptions

type ExchangeOptions struct {
	Name          string
	Type          ExchangeType
	IsDurable     bool
	IsAutoDeleted bool
	IsInternal    bool
	NoWait        bool
	Args          map[string]interface{}
}

type ExchangeType

type ExchangeType uint8

const (
	ExchangeTypeDirect ExchangeType = iota
	ExchangeTypeFanout
	ExchangeTypeTopic
	ExchangeTypeHeaders
)

func (ExchangeType) String

func (t ExchangeType) String() string

func (ExchangeType) Value

func (t ExchangeType) Value() ExchangeType

type Message

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

func NewMessage

func NewMessage() *Message

func (*Message) Ack

func (m *Message) Ack() error

func (*Message) GetAppID

func (m *Message) GetAppID() string

func (*Message) GetBody

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

func (*Message) GetContentEncoding

func (m *Message) GetContentEncoding() string

func (*Message) GetContentType

func (m *Message) GetContentType() string

func (*Message) GetCorrelationID

func (m *Message) GetCorrelationID() uuid.UUID

func (*Message) GetDeliveryMode

func (m *Message) GetDeliveryMode() uint8

func (*Message) GetExpiration

func (m *Message) GetExpiration() time.Duration

func (*Message) GetExpirationString

func (m *Message) GetExpirationString() string

func (*Message) GetHeader

func (m *Message) GetHeader(key string) interface{}

func (*Message) GetHeaders

func (m *Message) GetHeaders() map[string]interface{}

func (*Message) GetID

func (m *Message) GetID() uuid.UUID

func (*Message) GetPriority

func (m *Message) GetPriority() uint8

func (*Message) GetReplyTo

func (m *Message) GetReplyTo() string

func (*Message) GetTimestamp

func (m *Message) GetTimestamp() time.Time

func (*Message) GetType

func (m *Message) GetType() string

func (*Message) GetUserID

func (m *Message) GetUserID() string

func (*Message) Nack

func (m *Message) Nack() error

func (*Message) Reject

func (m *Message) Reject() error

func (*Message) SetAppID

func (m *Message) SetAppID(appID string) *Message

func (*Message) SetBody

func (m *Message) SetBody(body []byte) *Message

func (*Message) SetContentEncoding

func (m *Message) SetContentEncoding(v string) *Message

func (*Message) SetContentType

func (m *Message) SetContentType(v string) *Message

func (*Message) SetCorrelationID

func (m *Message) SetCorrelationID(id uuid.UUID) *Message

func (*Message) SetDeliveryModePersistent

func (m *Message) SetDeliveryModePersistent() *Message

func (*Message) SetExpiration

func (m *Message) SetExpiration(expiration time.Duration) *Message

func (*Message) SetHeader

func (m *Message) SetHeader(key string, value interface{}) *Message

func (*Message) SetPriority

func (m *Message) SetPriority(priority uint8) *Message

func (*Message) SetReplyTo

func (m *Message) SetReplyTo(v string) *Message

func (*Message) SetTimestamp

func (m *Message) SetTimestamp(timestamp time.Time) *Message

func (*Message) SetType

func (m *Message) SetType(v string) *Message

func (*Message) SetUserID

func (m *Message) SetUserID(useID string) *Message

type PublishOptions

type PublishOptions struct {
	Exchange    string
	RoutingKey  string
	IsMandatory bool
	IsImmediate bool
}

type Publisher

type Publisher interface {
	Publish(options *PublishOptions, message *Message) error
}

func NewPublisher

func NewPublisher(options *PublisherOptions) Publisher

type PublisherOptions

type PublisherOptions struct {
	URI string
}

type QueueOptions

type QueueOptions struct {
	Name          string
	Durable       bool
	AutoDelete    bool
	Exclusive     bool
	NoWait        bool
	MaxPriority   uint8
	RoutingKey    string
	QueueBindArgs map[string]interface{}
}

func (*QueueOptions) GetArgs

func (o *QueueOptions) GetArgs() amqp.Table

type Subscriber

type Subscriber interface {
	Subscribe(handler func(message *Message))
}

func NewSubscriber

func NewSubscriber(options *SubscriberOptions) Subscriber

type SubscriberOptions

type SubscriberOptions struct {
	URI             string
	QueueOptions    *QueueOptions
	ExchangeOptions *ExchangeOptions
	MaxPriority     uint8
	PrefetchCount   int
	Name            string
	AutoAck         bool
	NoWait          bool
	NoLocal         bool
	Exclusive       bool
	Args            map[string]interface{}
}