Refactoring

Author: elvin-tajirzada

README.md

@@ -1,6 +1,8 @@
-# Go RabbitMQ Consumer
+![GitHub go.mod Go version (subdirectory of monorepo)](https://img.shields.io/github/go-mod/go-version/elvin-tacirzade/go-amqp?logo=go)
 [![Go Reference](https://pkg.go.dev/badge/github.com/elvin-tacirzade/go-amqp.svg)](https://pkg.go.dev/github.com/elvin-tacirzade/go-amqp)
 
+# Go RabbitMQ Consumer
+
 This package uses the package provided by the [RabbitMQ core team](https://github.com/rabbitmq/amqp091-go).
 ## Goals
 Provide auto reconnect
@@ -9,14 +11,14 @@ Provide auto reconnect
 go get -u github.com/elvin-tacirzade/go-amqp
 ```
 ## Usage
-First we call the Init() function. The init() function takes the following parameters:
+First we call a New() function. The function takes the following parameters:
 1. `user` - Declare a RabbitMQ user.
 2. `password` - Declare a RabbitMQ password
 3. `host` - Declare a RabbitMQ host
 4. `port` -Declare a RabbitMQ port
 5. `reconnectTime` - Declare an auto reconnect time
 
-Init() function returns the RabbitMQ struct and error.
+New() function returns the RabbitMQ struct and error.
 
 See the [example](https://github.com/elvin-tacirzade/go-amqp/tree/main/example) subdirectory for simple consumers executables.
 

example/consumer.go

@@ -18,7 +18,7 @@ func main() {
 	prefetchCount := 5
 	var deliveries []string
 
-	rabbitMQ, err := amqp.Init(rabbitMQUser, rabbitMQPassword, rabbitMQHost, rabbitMQPort, reconnectTime)
+	rabbitMQ, err := amqp.New(rabbitMQUser, rabbitMQPassword, rabbitMQHost, rabbitMQPort, reconnectTime)
 	if err != nil {
 		log.Fatal(err)
 	}

rabbitmq.go

@@ -8,20 +8,21 @@ import (
 	"time"
 )
 
-// RabbitMQ struct defines Connection, Channel, queue, delay
+// RabbitMQ contains Connection and Channel
 type RabbitMQ struct {
 	Connection *amqp.Connection
 	Channel    *amqp.Channel
 	queue      amqp.Queue
+	dsn        string
 	delay      time.Duration
 	done       chan bool
 }
 
-// Init function connect to RabbitMQ and open a channel
-func Init(user, password, host, port string, reconnectTime time.Duration) (*RabbitMQ, error) {
-	url := fmt.Sprintf("amqp://%s:%s@%s:%s/", user, password, host, port)
+// New connects to RabbitMQ and open a channel
+func New(user, password, host, port string, reconnectTime time.Duration) (*RabbitMQ, error) {
+	dsn := fmt.Sprintf("amqp://%s:%s@%s:%s/", user, password, host, port)
 
-	conn, connErr := amqp.Dial(url)
+	conn, connErr := amqp.Dial(dsn)
 	if connErr != nil {
 		return nil, fmt.Errorf("failed to connect to RabbitMQ: %v", connErr)
 	}
@@ -32,15 +33,16 @@ func Init(user, password, host, port string, reconnectTime time.Duration) (*Rabb
 	}
 
 	done := make(chan bool)
-	rabbitMQ := &RabbitMQ{Connection: conn, Channel: ch, delay: reconnectTime, done: done}
 
-	go reconnectConnection(rabbitMQ, url)
-	go reconnectChannel(rabbitMQ)
+	rabbitMQ := &RabbitMQ{Connection: conn, Channel: ch, dsn: dsn, delay: reconnectTime, done: done}
+
+	rabbitMQ.reconnectConnection()
+	rabbitMQ.reconnectChannel()
 
 	return rabbitMQ, nil
 }
 
-// ExchangeDeclare function declare an exchange
+// ExchangeDeclare declare an exchange
 func (r *RabbitMQ) ExchangeDeclare(name, exchangeType string, durable, autoDelete, internal, noWait bool, arguments amqp.Table) error {
 	err := r.Channel.ExchangeDeclare(name, exchangeType, durable, autoDelete, internal, noWait, arguments)
 	if err != nil {
@@ -49,7 +51,7 @@ func (r *RabbitMQ) ExchangeDeclare(name, exchangeType string, durable, autoDelet
 	return nil
 }
 
-// QueueDeclare function declare a queue
+// QueueDeclare declare a queue
 func (r *RabbitMQ) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, arguments amqp.Table) error {
 	queue, queueErr := r.Channel.QueueDeclare(name, durable, autoDelete, exclusive, noWait, arguments)
 	if queueErr != nil {
@@ -59,7 +61,7 @@ func (r *RabbitMQ) QueueDeclare(name string, durable, autoDelete, exclusive, noW
 	return nil
 }
 
-// QueueBind function bind a queue
+// QueueBind bind a queue
 func (r *RabbitMQ) QueueBind(key, exchangeName string, noWait bool, arguments amqp.Table) error {
 	err := r.Channel.QueueBind(r.queue.Name, key, exchangeName, noWait, arguments)
 	if err != nil {
@@ -68,7 +70,7 @@ func (r *RabbitMQ) QueueBind(key, exchangeName string, noWait bool, arguments am
 	return nil
 }
 
-// Qos function controls how many messages or how many bytes the server will try to keep on
+// Qos controls how many messages or how many bytes the server will try to keep on
 // the network for consumers before receiving delivery ack
 func (r *RabbitMQ) Qos(prefetchCount, prefetchSize int, global bool) error {
 	err := r.Channel.Qos(prefetchCount, prefetchSize, global)
@@ -78,7 +80,7 @@ func (r *RabbitMQ) Qos(prefetchCount, prefetchSize int, global bool) error {
 	return nil
 }
 
-// Consume function starts delivering queued messages
+// Consume starts delivering queued messages
 func (r *RabbitMQ) Consume(consumer string, autoAck, exclusive, noLocal, noWait bool, arguments amqp.Table) (<-chan amqp.Delivery, error) {
 	deliveries := make(chan amqp.Delivery)
 
@@ -102,5 +104,4 @@ func (r *RabbitMQ) Consume(consumer string, autoAck, exclusive, noLocal, noWait
 	}()
 
 	return deliveries, nil
-
 }

reconnect.go

@@ -6,12 +6,12 @@ import (
 	"time"
 )
 
-// reconnectConnection function reconnect a connection
-func reconnectConnection(rabbitMQ *RabbitMQ, url string) {
+// reconnectConnection reconnect a connection
+func (r *RabbitMQ) reconnectConnection() {
 	for {
 		chanError := make(chan *amqp.Error)
 
-		closeReason, ok := <-rabbitMQ.Connection.NotifyClose(chanError)
+		closeReason, ok := <-r.Connection.NotifyClose(chanError)
 		if !ok {
 			log.Println("connection closed by developer")
 			break
@@ -20,11 +20,11 @@ func reconnectConnection(rabbitMQ *RabbitMQ, url string) {
 		log.Printf("reason for the connection closed: %v\n", closeReason)
 
 		for {
-			time.Sleep(rabbitMQ.delay)
+			time.Sleep(r.delay)
 
-			conn, connErr := amqp.Dial(url)
+			conn, connErr := amqp.Dial(r.dsn)
 			if connErr == nil {
-				rabbitMQ.Connection = conn
+				r.Connection = conn
 				log.Println("connection successfully reconnected")
 				break
 			}
@@ -34,29 +34,29 @@ func reconnectConnection(rabbitMQ *RabbitMQ, url string) {
 	}
 }
 
-// reconnectChannel function reconnect a channel
-func reconnectChannel(rabbitMQ *RabbitMQ) {
+// reconnectChannel reconnect a channel
+func (r *RabbitMQ) reconnectChannel() {
 	for {
 		chanError := make(chan *amqp.Error)
 
-		closeReason, ok := <-rabbitMQ.Channel.NotifyClose(chanError)
+		closeReason, ok := <-r.Channel.NotifyClose(chanError)
 
 		if !ok {
-			rabbitMQ.done <- true
+			r.done <- true
 			log.Println("channel closed by developer")
 			break
 		}
 
 		log.Printf("reason for the channel closed: %v\n", closeReason)
 
 		for {
-			time.Sleep(rabbitMQ.delay)
+			time.Sleep(r.delay)
 
-			ch, chErr := rabbitMQ.Connection.Channel()
+			ch, chErr := r.Connection.Channel()
 			if chErr == nil {
-				rabbitMQ.Channel = ch
+				r.Channel = ch
 				log.Println("channel successfully reconnected")
-				rabbitMQ.done <- false
+				r.done <- false
 				break
 			}