From 05e1e946d844f5676f27cde56d8c2af882424112 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 20 Sep 2025 14:45:53 +0000 Subject: [PATCH 1/2] Initial plan From 89842d2180169a28ccd467b5f77d891dcaac93cd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 20 Sep 2025 14:51:44 +0000 Subject: [PATCH 2/2] Implement complete Go RabbitMQ module with client, examples, and tests Co-authored-by: idprm <68960568+idprm@users.noreply.github.com> --- .gitignore | 22 +++++ README.md | 119 +++++++++++++++++++++++++++- example/consumer/consumer.go | 35 ++++++++ example/consumer/go.mod | 9 +++ example/consumer/go.sum | 4 + example/publisher/go.mod | 9 +++ example/publisher/go.sum | 4 + example/publisher/publisher.go | 41 ++++++++++ go.mod | 5 ++ go.sum | 4 + rmq.go | 141 +++++++++++++++++++++++++++++++++ rmq_test.go | 65 +++++++++++++++ 12 files changed, 457 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 example/consumer/consumer.go create mode 100644 example/consumer/go.mod create mode 100644 example/consumer/go.sum create mode 100644 example/publisher/go.mod create mode 100644 example/publisher/go.sum create mode 100644 example/publisher/publisher.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 rmq.go create mode 100644 rmq_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0c46795 --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +# Build artifacts +*.exe +*.exe~ +*.dll +*.so +*.dylib + +# Test binary, built with `go test -c` +*.test + +# Output of the go coverage tool, specifically when used with LiteIDE +*.out + +# Go workspace file +go.work + +# Example binaries +/example/publisher/publisher +/example/consumer/consumer + +# Temporary files +/tmp/ \ No newline at end of file diff --git a/README.md b/README.md index 7bbb4d6..a91e59d 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,119 @@ # go-rmq -RabbitMQ Driver + +A simple and easy-to-use Go RabbitMQ client library built on top of the official AMQP library. + +## Features + +- Simple connection management +- Easy message publishing +- Message consuming with custom handlers +- Automatic queue declaration +- Persistent message delivery +- Proper error handling and logging + +## Installation + +```bash +go get github.com/idprm/go-rmq +``` + +## Quick Start + +### Publishing Messages + +```go +package main + +import ( + "context" + "log" + + "github.com/idprm/go-rmq" +) + +func main() { + // Create client + client, err := rmq.NewClient(rmq.Config{ + URL: "amqp://guest:guest@localhost:5672/", + }) + if err != nil { + log.Fatal(err) + } + defer client.Close() + + // Publish message + ctx := context.Background() + err = client.PublishMessage(ctx, "my_queue", []byte("Hello, RabbitMQ!")) + if err != nil { + log.Fatal(err) + } +} +``` + +### Consuming Messages + +```go +package main + +import ( + "fmt" + "log" + + "github.com/idprm/go-rmq" +) + +func main() { + // Create client + client, err := rmq.NewClient(rmq.Config{ + URL: "amqp://guest:guest@localhost:5672/", + }) + if err != nil { + log.Fatal(err) + } + defer client.Close() + + // Define message handler + handler := func(message []byte) error { + fmt.Printf("Received: %s\n", string(message)) + return nil + } + + // Start consuming (blocking) + err = client.ConsumeMessages("my_queue", handler) + if err != nil { + log.Fatal(err) + } +} +``` + +## Configuration + +The `Config` struct supports the following options: + +- `URL`: RabbitMQ connection URL (default: "amqp://guest:guest@localhost:5672/") + +## API Reference + +### Client Methods + +- `NewClient(config Config) (*Client, error)` - Create a new RabbitMQ client +- `Close() error` - Close the connection +- `DeclareQueue(name string) error` - Declare a queue +- `PublishMessage(ctx context.Context, queueName string, message []byte) error` - Publish a message +- `ConsumeMessages(queueName string, handler func([]byte) error) error` - Consume messages +- `IsConnected() bool` - Check connection status + +## Examples + +See the `example/` directory for complete working examples: +- `example/publisher/` - Message publisher example +- `example/consumer/` - Message consumer example + +## Requirements + +- Go 1.18+ +- RabbitMQ server + +## License + +MIT License diff --git a/example/consumer/consumer.go b/example/consumer/consumer.go new file mode 100644 index 0000000..8afd0da --- /dev/null +++ b/example/consumer/consumer.go @@ -0,0 +1,35 @@ +package main + +import ( + "fmt" + "log" + + "github.com/idprm/go-rmq" +) + +func main() { + // Create RabbitMQ client + client, err := rmq.NewClient(rmq.Config{ + URL: "amqp://guest:guest@localhost:5672/", + }) + if err != nil { + log.Fatalf("Failed to create RabbitMQ client: %v", err) + } + defer client.Close() + + // Queue name + queueName := "test_queue" + + // Message handler + handler := func(message []byte) error { + fmt.Printf("Received message: %s\n", string(message)) + return nil + } + + // Start consuming messages + fmt.Printf("Starting to consume messages from queue '%s'...\n", queueName) + err = client.ConsumeMessages(queueName, handler) + if err != nil { + log.Fatalf("Failed to consume messages: %v", err) + } +} diff --git a/example/consumer/go.mod b/example/consumer/go.mod new file mode 100644 index 0000000..74a0996 --- /dev/null +++ b/example/consumer/go.mod @@ -0,0 +1,9 @@ +module consumer + +go 1.24.7 + +require github.com/idprm/go-rmq v0.0.0 + +require github.com/rabbitmq/amqp091-go v1.10.0 // indirect + +replace github.com/idprm/go-rmq => ../../ diff --git a/example/consumer/go.sum b/example/consumer/go.sum new file mode 100644 index 0000000..024eebe --- /dev/null +++ b/example/consumer/go.sum @@ -0,0 +1,4 @@ +github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw= +github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= diff --git a/example/publisher/go.mod b/example/publisher/go.mod new file mode 100644 index 0000000..e327d59 --- /dev/null +++ b/example/publisher/go.mod @@ -0,0 +1,9 @@ +module publisher + +go 1.24.7 + +require github.com/idprm/go-rmq v0.0.0 + +require github.com/rabbitmq/amqp091-go v1.10.0 // indirect + +replace github.com/idprm/go-rmq => ../../ diff --git a/example/publisher/go.sum b/example/publisher/go.sum new file mode 100644 index 0000000..024eebe --- /dev/null +++ b/example/publisher/go.sum @@ -0,0 +1,4 @@ +github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw= +github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= diff --git a/example/publisher/publisher.go b/example/publisher/publisher.go new file mode 100644 index 0000000..6f2a93e --- /dev/null +++ b/example/publisher/publisher.go @@ -0,0 +1,41 @@ +package main + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/idprm/go-rmq" +) + +func main() { + // Create RabbitMQ client + client, err := rmq.NewClient(rmq.Config{ + URL: "amqp://guest:guest@localhost:5672/", + }) + if err != nil { + log.Fatalf("Failed to create RabbitMQ client: %v", err) + } + defer client.Close() + + // Queue name + queueName := "test_queue" + + // Publish messages + ctx := context.Background() + for i := 0; i < 5; i++ { + message := fmt.Sprintf("Hello RabbitMQ! Message #%d", i+1) + + err := client.PublishMessage(ctx, queueName, []byte(message)) + if err != nil { + log.Printf("Failed to publish message: %v", err) + continue + } + + fmt.Printf("Published: %s\n", message) + time.Sleep(1 * time.Second) + } + + fmt.Println("All messages published successfully!") +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..9d11301 --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module github.com/idprm/go-rmq + +go 1.24.7 + +require github.com/rabbitmq/amqp091-go v1.10.0 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..024eebe --- /dev/null +++ b/go.sum @@ -0,0 +1,4 @@ +github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw= +github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= diff --git a/rmq.go b/rmq.go new file mode 100644 index 0000000..25e59cc --- /dev/null +++ b/rmq.go @@ -0,0 +1,141 @@ +package rmq + +import ( + "context" + "fmt" + "log" + "time" + + amqp "github.com/rabbitmq/amqp091-go" +) + +// Client represents a RabbitMQ client +type Client struct { + conn *amqp.Connection + channel *amqp.Channel + url string +} + +// Config holds configuration for RabbitMQ connection +type Config struct { + URL string +} + +// NewClient creates a new RabbitMQ client +func NewClient(config Config) (*Client, error) { + if config.URL == "" { + config.URL = "amqp://guest:guest@localhost:5672/" + } + + conn, err := amqp.Dial(config.URL) + if err != nil { + return nil, fmt.Errorf("failed to connect to RabbitMQ: %w", err) + } + + channel, err := conn.Channel() + if err != nil { + conn.Close() + return nil, fmt.Errorf("failed to open channel: %w", err) + } + + return &Client{ + conn: conn, + channel: channel, + url: config.URL, + }, nil +} + +// Close closes the RabbitMQ connection +func (c *Client) Close() error { + if c.channel != nil { + if err := c.channel.Close(); err != nil { + log.Printf("Error closing channel: %v", err) + } + } + if c.conn != nil { + if err := c.conn.Close(); err != nil { + log.Printf("Error closing connection: %v", err) + return err + } + } + return nil +} + +// DeclareQueue declares a queue +func (c *Client) DeclareQueue(name string) error { + _, err := c.channel.QueueDeclare( + name, // queue name + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments + ) + return err +} + +// PublishMessage publishes a message to a queue +func (c *Client) PublishMessage(ctx context.Context, queueName string, message []byte) error { + // Ensure queue exists + if err := c.DeclareQueue(queueName); err != nil { + return fmt.Errorf("failed to declare queue: %w", err) + } + + return c.channel.PublishWithContext( + ctx, + "", // exchange + queueName, // routing key + false, // mandatory + false, // immediate + amqp.Publishing{ + ContentType: "text/plain", + Body: message, + DeliveryMode: amqp.Persistent, // make message persistent + Timestamp: time.Now(), + }, + ) +} + +// ConsumeMessages consumes messages from a queue +func (c *Client) ConsumeMessages(queueName string, handler func([]byte) error) error { + // Ensure queue exists + if err := c.DeclareQueue(queueName); err != nil { + return fmt.Errorf("failed to declare queue: %w", err) + } + + msgs, err := c.channel.Consume( + queueName, // queue + "", // consumer + false, // auto-ack + false, // exclusive + false, // no-local + false, // no-wait + nil, // args + ) + if err != nil { + return fmt.Errorf("failed to register consumer: %w", err) + } + + forever := make(chan bool) + + go func() { + for d := range msgs { + if err := handler(d.Body); err != nil { + log.Printf("Error handling message: %v", err) + d.Nack(false, true) // negative acknowledge, requeue + } else { + d.Ack(false) // acknowledge message + } + } + }() + + log.Printf("Waiting for messages from queue '%s'. To exit press CTRL+C", queueName) + <-forever + + return nil +} + +// IsConnected checks if the client is connected to RabbitMQ +func (c *Client) IsConnected() bool { + return c.conn != nil && !c.conn.IsClosed() +} diff --git a/rmq_test.go b/rmq_test.go new file mode 100644 index 0000000..021adef --- /dev/null +++ b/rmq_test.go @@ -0,0 +1,65 @@ +package rmq + +import ( + "testing" +) + +func TestNewClient(t *testing.T) { + // Test with default config + config := Config{} + client, err := NewClient(config) + + // We expect this to fail if RabbitMQ is not running + // which is normal in a test environment + if err != nil { + t.Logf("Expected error when RabbitMQ is not available: %v", err) + return + } + + if client == nil { + t.Error("Expected client to be created") + return + } + + defer client.Close() + + // Test connection status + if !client.IsConnected() { + t.Error("Expected client to be connected") + } +} + +func TestClientClose(t *testing.T) { + config := Config{ + URL: "amqp://invalid:invalid@nonexistent:5672/", + } + + client, err := NewClient(config) + if err != nil { + // Expected to fail with invalid connection + t.Logf("Expected error with invalid config: %v", err) + return + } + + if client != nil { + err = client.Close() + if err != nil { + t.Errorf("Error closing connection: %v", err) + } + } +} + +func TestConfig(t *testing.T) { + // Test default URL + config := Config{} + if config.URL != "" { + t.Error("Expected empty URL in default config") + } + + // Test custom URL + customURL := "amqp://user:pass@host:5672/vhost" + config = Config{URL: customURL} + if config.URL != customURL { + t.Errorf("Expected URL %s, got %s", customURL, config.URL) + } +}