rabbitmq

README

rabbitmq-client

Documentation

Index

• Variables
• func FailOnError(err error, msg string)
• func StartConsumer(destination *MqDestination, receiver OnReceive, connSetting *Settings)
• func WrapRepo(msg amqp.Delivery, body []byte, headers amqp.Table) *amqp.Publishing
• type MqDestination
  • func (mq *MqDestination) Consume(conn *rabbitmq.Connection, consumerTag string) (<-chan amqp.Delivery, *rabbitmq.Channel, error)
  • func (mq *MqDestination) DeclareDestination(cnn *rabbitmq.Connection, createTempQueue bool) error
  • func (mq *MqDestination) Produce(channel *rabbitmq.Channel, message amqp.Publishing) error
  • func (mq *MqDestination) RPC(ctx context.Context, conn *rabbitmq.Connection, message amqp.Publishing) (*amqp.Delivery, error)
• type OnReceive
• type Settings
  • func (r *Settings) ConnURL() string
  • func (r *Settings) Connect() (*rabbitmq.Connection, error)
  • func (r *Settings) String() string

Variables

var SharedConnection = true

SharedConnection should shared connection? default YES

Functions

func FailOnError

func FailOnError(err error, msg string)

FailOnError failed if any error

func StartConsumer

func StartConsumer(destination *MqDestination, receiver OnReceive, connSetting *Settings)

StartConsumer start process.

func WrapRepo

func WrapRepo(msg amqp.Delivery, body []byte, headers amqp.Table) *amqp.Publishing

WrapRepo wrap up for repo to rabbitmq

Types

type MqDestination

type MqDestination struct {
	Queue        string
	Topic        string
	ExchangeType string
	AutoAck      bool
	Exclusive    bool
	Prefetch     int
	DeclareAll   bool
}

MqDestination Rabbitmq destination

func (*MqDestination) Consume

func (mq *MqDestination) Consume(conn *rabbitmq.Connection, consumerTag string) (<-chan amqp.Delivery, *rabbitmq.Channel, error)

Consume start consumer

func (*MqDestination) DeclareDestination

func (mq *MqDestination) DeclareDestination(cnn *rabbitmq.Connection, createTempQueue bool) error

DeclareDestination declare Topic, queues....

func (*MqDestination) Produce

func (mq *MqDestination) Produce(channel *rabbitmq.Channel, message amqp.Publishing) error

Produce publish message

func (*MqDestination) RPC

func (mq *MqDestination) RPC(ctx context.Context, conn *rabbitmq.Connection, message amqp.Publishing) (*amqp.Delivery, error)

RPC RPC over rabbitmq message. timeout setting should be ctx

type OnReceive

type OnReceive interface {
	OnReceiveMessage(msg amqp.Delivery) (string, *amqp.Publishing, error)
}

OnReceive interface for Receiver

type Settings

type Settings struct {
	Host     string `default:"localhost"`
	Port     uint   `default:"5672"`
	User     string `default:"guest"`
	Password string `default:"guest"`
	Vhost    string `default:"/"`
	Prop     amqp.Table
	// contains filtered or unexported fields
}

Settings Settings, should include url & options

func (*Settings) ConnURL

func (r *Settings) ConnURL() string

ConnURL return connection URL for Dial

func (*Settings) Connect

func (r *Settings) Connect() (*rabbitmq.Connection, error)

Connect make connection to Rabbitmq

func (*Settings) String

func (r *Settings) String() string

String for log connection URL.