eqmrmq

README

EQMRMQ

EQMRMQ is a Go package for simplified interaction with RabbitMQ, specifically designed for sending messages, receiving responses, and consuming messages from queues with ease.

The eqmrmq package utilizes the github.com/rabbitmq/amqp091-go package, which is an AMQP 0.9.1 Go client library. This library provides the underlying functionality for interacting with RabbitMQ, including features such as establishing connections, creating channels, publishing messages, consuming messages, and handling acknowledgments. By leveraging amqp091-go, eqmrmq simplifies the process of sending and receiving messages to and from RabbitMQ queues within Go applications. EQMRMQ utilizes the slog package for structured logging.

Installation

To install EQMRMQ, use go get:

go get github.com/lacolle87/eqmrmq

Usage

Import EQMRMQ into your Go project:

import "github.com/yourusername/eqmrmq"

Establishing a Connection

To establish a connection to RabbitMQ and monitor it:

package main

import (
	"log"
	"github.com/lacolle87/eqmrmq"
)

func main() {
	rabbitURL := "amqp://guest:guest@localhost:5672/"
	conn, err := eqmrmq.Connect(rabbitURL)
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
	}
	defer conn.Close()
}

The Connect function establishes a connection to RabbitMQ and starts monitoring it to automatically reconnect if the connection is lost.

Publishing a Message

To Publish a message to a queue:

func main() {
    // Create a new RabbitMQ connection
    rabbitURL: = "amqp://guest:guest@localhost:5672/"
    conn,
    err: = eqmrmq.Connect(rabbitURL)
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    // Create a channel
    ch,
    err: = conn.Channel()
    if err != nil {
        panic(err)
    }
    defer ch.Close()

    // Generate a correlation ID
    correlationId: = eqmrmq.GenerateCorrelationId()

    // Send a message
        msg: = eqmrmq.Message {
        QueueName: "my_queue",
        Message: "Hello, RabbitMQ!",
        CorrelationId: correlationId,
        ReplyQueue: "", // Assuming no reply queue is needed here
        Ch: ch,
    }

        err = msg.Publish()
    if err != nil {
        log.Fatalf("Failed to publish message: %v", err)
    }
}

Publishing a Message with Response

To send a message to a queue and wait for a response:

// Send a message and wait for response
response, err := eqmrmq.PublishToQueueWithResponse("my_queue", "Hello, RabbitMQ!", ch)
if err != nil {
	panic(err)
}
fmt.Println("Response:", string(response))

Consuming Messages

To consume messages from a queue:

// Define a handler function
handler := func(ch *amqp.Channel, d amqp.Delivery, arg interface{}) error {
    fmt.Println("Received message:", string(d.Body))
    return nil
}

// Any additional argument you want to pass to the handler
additionalArg := "my_custom_argument" // This can be any type

// Consume messages
err := eqmrmq.ConsumeMessages(ch, "my_queue", handler, additionalArg)
if err != nil {
    panic(err)
}

Replying to a Message

To reply to a message:

// Reply to a message
err := eqmrmq.ReplyToMessage(ch, delivery, []byte("Response from server"))
if err != nil {
	panic(err)
}

Acknowledgments

Special thanks to the authors of RabbitMQ and the AMQP 0.9.1 Go client library github.com/rabbitmq/amqp091-go for providing the underlying functionality used by this package!

Documentation

Index

• Constants
• func Connect(rabbitURL string) (*amqp.Connection, error)
• func ConsumeMessages(ch *amqp.Channel, queueName string, ...) error
• func CreateReplyQueue(channel *amqp.Channel) (string, error)
• func DeclareQueue(ch *amqp.Channel, queueName string) (amqp.Queue, error)
• func GenerateCorrelationId() string
• func PublishToQueueWithResponse(msg Message) ([]byte, error)
• func ReceiveResponse(correlationId, replyQueue string, ch *amqp.Channel) ([]byte, error)
• func RegisterConsumer(ch *amqp.Channel, queueName string) (<-chan amqp.Delivery, error)
• func ReplyToMessage(ch *amqp.Channel, d amqp.Delivery, replyData []byte) error
• type Message
  • func (msg Message) Publish() error

Constants

const (
	ContentTypeTextPlain contentType = "text/plain"
	ContentTypeTextHTML  contentType = "text/html"
	ContentTypeTextCSV   contentType = "text/csv"

	ContentTypeJSON contentType = "application/json"
	ContentTypeXML  contentType = "application/xml"

	ContentTypeOctetStream contentType = "application/octet-stream"
	ContentTypePDF         contentType = "application/pdf"
	ContentTypeJPEG        contentType = "image/jpeg"
	ContentTypePNG         contentType = "image/png"
	ContentTypeGIF         contentType = "image/gif"
	ContentTypeAVIF        contentType = "image/avif"
	ContentTypeZIP         contentType = "application/zip"
)

Functions

func Connect

func Connect(rabbitURL string) (*amqp.Connection, error)

func ConsumeMessages

func ConsumeMessages(ch *amqp.Channel, queueName string, handler func(*amqp.Channel, amqp.Delivery, interface{}) error, arg interface{}) error

func CreateReplyQueue

func CreateReplyQueue(channel *amqp.Channel) (string, error)

func DeclareQueue

func DeclareQueue(ch *amqp.Channel, queueName string) (amqp.Queue, error)

func GenerateCorrelationId

func GenerateCorrelationId() string

func PublishToQueueWithResponse

func PublishToQueueWithResponse(msg Message) ([]byte, error)

func ReceiveResponse

func ReceiveResponse(correlationId, replyQueue string, ch *amqp.Channel) ([]byte, error)

func RegisterConsumer

func RegisterConsumer(ch *amqp.Channel, queueName string) (<-chan amqp.Delivery, error)

func ReplyToMessage

func ReplyToMessage(ch *amqp.Channel, d amqp.Delivery, replyData []byte) error

Types

type Message

type Message struct {
	QueueName     string
	ContentType   contentType
	Message       []byte
	CorrelationId string
	ReplyQueue    string
	Ch            *amqp.Channel
}

func (Message) Publish

func (msg Message) Publish() error