rmqgo

README

🐰 rmqgo

Wrapper of rabbitmq/amqp091-go that provides some features.

Installation

go get github.com/Almazatun/rmqgo

Connect to rabbitMQ

import (rmqgo "github.com/Almazatun/rmqgo")

rmq := rmqgo.New()

config := rmqgo.ConnectConfig{
 User: "user",
 Pass: "pass",
 Host: "host",
 Port: "port",
}

err := rmq.Connect(config)

if err != nil {
 // some action
}

Optional params when initialize rmqgo.New()

With RPC mode for request and replay pattern

rmqgo.New(rmqgo.WithRpc(replayQueueName, exchangeType))

With topic RPC

rmqgo.New(rmqgo.WithTopicRpc(replayQueueName, exchangeType, routingKey))

Create channel

ch, err := rmq.CreateChannel()

if err != nil {
 // some action
}

Create queue

args := make(map[string]interface{})

q, err := rmq.CreateQueue(rmqgo.CreateQueueConfig{
 Name:         "some_name",
 DeleteUnused: false,
 Exclusive:    false,
 NoWait:       false,
 Durable:      true,
 Args:         &args,
})

if err != nil {
 // some action
}

Create exchange

Exchanges

import (rmqgo "github.com/Almazatun/rmqgo")

rmqgo.Exchanges.Direct()
rmqgo.Exchanges.Topic()
rmqgo.Exchanges.Fanout()
rmqgo.Exchanges.Headers()

Exchange types

import (rmqgo "github.com/Almazatun/rmqgo")

rmqgo.ExchangeType.Direct()
rmqgo.ExchangeType.Topic()
rmqgo.ExchangeType.Fanout()

args := make(map[string]interface{})

err := rmq.CreateExchange(rmqgo.CreateExchangeConfig{
 Name:       rmqgo.Exchanges.RmqDirect,
 Type:       rmqgo.ExchangeType.Direct,
 Durable:    true,
 AutoDelete: false,
 Internal:   false,
 NoWait:     false,
 Args:       &args,
})

if err != nil {
 // some action
}

Bind exchange by created queue

args := make(map[string]interface{})

err := rmq.BindQueueByExchange(rmqgo.BindQueueByExgConfig{
 QueueName:    "some_name",
 RoutingKey:   "some_key",
 ExchangeName: Exchanges.RmqDirect,
 NoWait:       false,
 Args:         &args,
})

if err != nil {
 // some action
}

Create producer

producer = rmqgo.NewProducer(&rmq)

Send message

err := producer.Send(Exchanges.RmqDirect, routingKey, msg, method)

if err != nil {
 // some action
}

Send message with reply

b, err := producer.SendReply(Exchanges.RmqDirect, routingKey, msg, method)

if err != nil {
 // some action
}

// msg - is your own type SomeName struct { someFields:... }

err = json.Unmarshal(*b, &msg)

if err != nil {
 // some action
}

Create consumer

consumer := rmqgo.NewConsumer(
  &rmq,
  rmqgo.WithConsumerConfig(rmqgo.CreateConsumerConfig{
   NameQueue: "some_name",
   Consumer:  "some_value",
   AutoAck:   false,
   Exclusive: false,
   NoWait:    false,
   NoLocal:   false,
  }),
 )

consumer.Listen()

Consuming messages from queues

// Bytes - <- chan []byte
<- rmq.ReceiveMessages()

consumer.Listen()

Optional params when initialize rmqgo.NewConsumer(...)

With HttpConsumer

rmqgo.NewConsumer(*rmq, rmqgo.WithHttpConsumer())

With Consumer Args

rmqgo.NewConsumer(*rmq, rmqgo.WithConsumerArgs(rmqgo.ConsumerArgs{
 XDeadLetterExc        *""
 XDeadLetterRoutingKey *""
 Ttl                   *int
 XExpires              *int
 XMaxPriority          *int
}))

Documentation

Index

• Variables
• func WithConsumerArgs(config ConsumerArgs) consumerOption
• func WithConsumerConfig(config CreateConsumerConfig) consumerOption
• func WithHttpConsumer() consumerOption
• type BindQueueByExgConfig
• type ConnectConfig
• type Consumer
  • func NewConsumer(rmq *Rmq, options ...consumerOption) *Consumer
  • func (c *Consumer) AddTopicsFuncs(topicsFuncs map[string]func([]byte) interface{}) error
  • func (c *Consumer) Listen()
• type ConsumerArgs
• type CreateConsumerConfig
• type CreateExchangeConfig
• type CreateQueueConfig
• type Msg
• type Producer
  • func NewProducer(rmq *Rmq, options ...producerOption) *Producer
  • func (p *Producer) Send(ex, rk string, msg interface{}, method string) error
  • func (p *Producer) SendReply(ex, rk string, msg interface{}, method string) (res []byte, err error)
• type ProducerInitConfig
• type Rmq
  • func New(options ...RmqOption) *Rmq
  • func (rmq *Rmq) BindQueueByExchange(config BindQueueByExgConfig) error
  • func (rmq *Rmq) Close() error
  • func (rmq *Rmq) Connect(config ConnectConfig) error
  • func (rmq *Rmq) CreateChannel() (c *amqp.Channel, err error)
  • func (rmq *Rmq) CreateExchange(config CreateExchangeConfig) error
  • func (rmq *Rmq) CreateQueue(config CreateQueueConfig) (q *amqp.Queue, err error)
  • func (rmq *Rmq) ReceiveMessages() <-chan []byte
• type RmqOption
  • func WithRpc(replayQueueName, exchangeType string) RmqOption
  • func WithTopicRpc(replayQueueName, exchangeType, rk string) RmqOption
• type SendMsg

Variables

var ExchangeType = exchangeType{}

var Exchanges = exchanges{}

Functions

func WithConsumerArgs

func WithConsumerArgs(config ConsumerArgs) consumerOption

Consumer

func WithConsumerConfig

func WithConsumerConfig(config CreateConsumerConfig) consumerOption

func WithHttpConsumer

func WithHttpConsumer() consumerOption

Make able to run in other thread when init Consumer It can be used if need to run rmq service with http

Types

type BindQueueByExgConfig

type BindQueueByExgConfig struct {
	QueueName    string
	RoutingKey   string
	ExchangeName string
	NoWait       bool
	Args         *map[string]interface{}
}

type ConnectConfig

type ConnectConfig struct {
	User string
	Pass string
	Host string
	Port string
}

type Consumer

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

func NewConsumer

func NewConsumer(rmq *Rmq, options ...consumerOption) *Consumer

func (*Consumer) AddTopicsFuncs

func (c *Consumer) AddTopicsFuncs(topicsFuncs map[string]func([]byte) interface{}) error

func (*Consumer) Listen

func (c *Consumer) Listen()

type ConsumerArgs

type ConsumerArgs struct {
	XDeadLetterExc        *string
	XDeadLetterRoutingKey *string
	Ttl                   *int
	XExpires              *int
	XMaxPriority          *int
}

type CreateConsumerConfig

type CreateConsumerConfig struct {
	NameQueue string
	Consumer  string
	AutoAck   bool
	Exclusive bool
	NoWait    bool
	NoLocal   bool
}

type CreateExchangeConfig

type CreateExchangeConfig struct {
	Name       string
	Type       string
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Args       *map[string]interface{}
}

type CreateQueueConfig

type CreateQueueConfig struct {
	// https://www.rabbitmq.com/ttl.html
	// By default 30_000 millisecond
	MsgTtl       *int
	Name         string //queue name
	DeleteUnused bool   //delete when unused
	Exclusive    bool
	NoWait       bool
	Durable      bool
	Args         *map[string]interface{}
}

type Msg

type Msg interface{}

type Producer

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

func NewProducer

func NewProducer(rmq *Rmq, options ...producerOption) *Producer

func (*Producer) Send

func (p *Producer) Send(ex, rk string, msg interface{}, method string) error

func (*Producer) SendReply

func (p *Producer) SendReply(ex, rk string, msg interface{}, method string) (res []byte, err error)

type ProducerInitConfig

type ProducerInitConfig struct {
	NameQueue    string
	ExchangeName string
}

type Rmq

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

func New

func New(options ...RmqOption) *Rmq

func (*Rmq) BindQueueByExchange

func (rmq *Rmq) BindQueueByExchange(config BindQueueByExgConfig) error

func (*Rmq) Close

func (rmq *Rmq) Close() error

func (*Rmq) Connect

func (rmq *Rmq) Connect(config ConnectConfig) error

func (*Rmq) CreateChannel

func (rmq *Rmq) CreateChannel() (c *amqp.Channel, err error)

func (*Rmq) CreateExchange

func (rmq *Rmq) CreateExchange(config CreateExchangeConfig) error

func (*Rmq) CreateQueue

func (rmq *Rmq) CreateQueue(config CreateQueueConfig) (q *amqp.Queue, err error)

func (*Rmq) ReceiveMessages

func (rmq *Rmq) ReceiveMessages() <-chan []byte

Only read access

type RmqOption

type RmqOption func(*Rmq)

func WithRpc

func WithRpc(replayQueueName, exchangeType string) RmqOption

Optionals

func WithTopicRpc

func WithTopicRpc(replayQueueName, exchangeType, rk string) RmqOption

type SendMsg

type SendMsg struct {
	Method string
	Msg
}