rmq

A simple wrapper for Go RabbitMQ client http://github.com/streadway/amqp Minimizes boilerplate code for using with dockerized RabbitMQ server

example

package main

import (
	"github.com/kaxap/rmq"
	"github.com/streadway/amqp"
	"log"
)

type something struct {
	ID   int    `json:"id"`
	Data string `json:"data"`
}

func main() {

	// connect to the RabbitMQ server
	// connection parameters should be present as environment variables
	// i.e. RABBITMQ_DEFAULT_USER, RABBITMQ_DEFAULT_PASS, RABBITMQ_HOST, RABBITMQ_PORT
	// for more information see https://hub.docker.com/_/rabbitmq/

	// this is short syntax for a durable consumer queue,
	// if you need to create a non-durable queue, please use NewQueue constructor (see "Constructors" chapter below)
	inputQueue, err := rmq.NewConsumerQueue("input_queue", 1)
	if err != nil {
		// could not connect or create channel
		log.Fatal(err)
	}
	defer inputQueue.Close()

	// this is short syntax for a durable producer queue
	outputQueue, err := rmq.NewProducerQueue("output_queue")
	if err != nil {
		// could not connect or create channel
		log.Fatal(err)
	}
	defer outputQueue.Close()

	// change delivery mode for Send and PublishJSON methods
	outputQueue.DeliveryMode = amqp.Persistent

	var a something

	// consume a json encoded message
	msg, err := inputQueue.Get(&a)
	log.Printf("message = %s\n", a.Data)

	// acknowledge the message
	msg.Ack(false)

	// modify data
	a.ID ++
	a.Data += " to you too"

	// send it to output queue
	outputQueue.Send(&a)
}

Now publish {"id": 1, "data": "hello"} to the "input_queue" and see what happens.

Even shorter syntax if you don't need to send messages very often

package main

import (
	"github.com/kaxap/rmq"
	"log"
)

type something struct {
	ID   int    `json:"id"`
	Data string `json:"data"`
}

func main() {

    const maxRetries := 10
    data := &something{ID: 1, Data: "hello!"}
    rmq.SendAndForget("some_queue", data, maxRetries)
    // this will try to connect to a durable queue with the given name and send the data 
    // if the connection or sending has failed it will retry up to 10 times. Then it will close the connection

Note if you need to send messages to non durable queues, please use rmq.SendAndForgetNonDurable. There is also SendAndForgetLazy available for lazy queues.

Constructors

There is 6 types of queue constructors:

func NewQueue(
    name string, // queue name
    durable bool, // durable flag
    prefetchCount int, // how many message to prefetch for this client
    autoAck bool, // auto_ack flag
    consume bool, // true for consumer/producer worker, false for producer-only worker
    autoReconnect bool, // true for auto reconnect to rabbitmq server
) (*Queue, error)

func NewQueueWithArgs(
    name string, // queue name
    durable bool, // durable flag
    prefetchCount int, // how many message to prefetch for this client
    autoAck bool, // auto_ack flag
    consume bool, // true for consumer/producer worker, false for producer-only worker
    autoReconnect bool, // true for auto reconnect to rabbitmq server
    args amqp.Table, // map[string]interface{} of queue arguments
) (*Queue, error)

NewProducerQueue(name string) (*Queue, error)
// short syntax for NewQueue(name, durable=true, prefetchCount=0, autoAck=false, consume=false, autoReconnect=true)

NewConsumerQueue(name string, prefetchCount int) (*Queue, error)
// short syntax for NewQueue(name, durable=true, prefetchCount=prefetchCount, autoAck=false, consume=true, autoReconnect=true)

NewLazyProducerQueue(name string) (*Queue, error)
// short syntax for producer queue with x-queue-mode: lazy args

NewLazyConsumerQueue(name string, prefetchCount int) (*Queue, error)
// short syntax for consumer queue with x-queue-mode: lazy args

Note that lazy queues are often used when a queue is expected to be frequently flooded. In lazy mode RabbitMQ pages out the messages on disk when possible. For more information see https://www.rabbitmq.com/lazy-queues.html