Refactor code

Author: minhduc140583

README.md

@@ -1,15 +1,78 @@
 # sqs
+A fully managed message queue service offered by AWS. It provides a reliable, scalable, and cost-effective way to decouple and coordinate distributed software systems and microservices.
+
+### Libraries for Amazon SQS (Simple Queue Service)
+- GO: [sqs](https://github.com/core-go/sqs), to wrap and simplify [aws-sdk-go/service/sqs](https://github.com/aws/aws-sdk-go/tree/main/service/sqs). Example is at [go-amazon-sqs-sample](https://github.com/project-samples/go-amazon-sqs-sample)
+
+#### A common flow to consume a message from a message queue
+![A common flow to consume a message from a message queue](https://cdn-images-1.medium.com/max/800/1*Y4QUN6QnfmJgaKigcNHbQA.png)
+- The libraries to implement this flow are:
+  - [mq](https://github.com/core-go/mq) for GOLANG. Example is at [go-amazon-sqs-sample](https://github.com/project-samples/go-amazon-sqs-sample)
+
+### Use Cases of Amazon SQS (Simple Queue Service)
+![Microservice Architecture](https://cdn-images-1.medium.com/max/800/1*vKeePO_UC73i7tfymSmYNA.png)
+#### Decoupling Microservices
+- <b>Scenario</b>: Separating different parts of an application to ensure that a failure in one part does not affect others.
+- <b>Benefit</b>: Enhances fault tolerance and scalability by allowing asynchronous communication between services.
+#### Asynchronous Processing
+- <b>Scenario</b>: Handling tasks that do not need immediate processing, such as batch processing or background tasks.
+- <b>Benefit</b>: Improves system efficiency and response times for end-users.
+  ![A typical micro service](https://cdn-images-1.medium.com/max/800/1*d9kyekAbQYBxH-C6w38XZQ.png)
+#### Job Queuing
+- <b>Scenario</b>: Managing and distributing jobs to worker processes.
+- <b>Benefit</b>: Balances load and ensures all tasks are completed without overloading any single worker.
+#### Order Processing Systems
+- <b>Scenario</b>: Processing customer orders, where each order can be handled as a separate task.
+- <b>Benefit</b>: Ensures reliable and scalable processing of orders, even during high demand.
+#### Message Buffering
+- <b>Scenario</b>: Smoothing out bursty traffic in applications to prevent overload.
+- <b>Benefit</b>: Protects the system from spikes in traffic by buffering messages.
+#### Workflow Orchestration
+- <b>Scenario</b>: Orchestrating steps in a complex workflow, such as image processing pipelines.
+- <b>Benefit</b>: Coordinates different stages of processing in a reliable and scalable manner.
+
+## Comparison of Amazon SQS, Google Pub/Sub and Apache Kafka
+#### Amazon SQS
+- <b>Type</b>: Managed message queuing service.
+- <b>Use Case</b>: Decoupling and scaling microservices, asynchronous tasks.
+- <b>Scalability</b>: Automatically scales.
+- <b>Delivery Guarantees</b>: At-least-once, FIFO (exactly-once).
+- <b>Integration</b>: Deep integration with AWS services.
+- <b>Delivery Models</b>: Primarily pull, with long polling.
+
+#### Google Pub/Sub:
+- <b>Type</b>: Managed real-time messaging service.
+- <b>Use Case</b>: Event-driven architectures, real-time analytics.
+- <b>Scalability</b>: Automatically scales.
+- <b>Delivery Guarantees</b>: At-least-once delivery.
+- <b>Integration</b>: Tight with Google Cloud services.
+- <b>Delivery Models</b>: Push and pull.
+
+#### Apache Kafka
+- <b>Type</b>: Open-source event streaming platform.
+- <b>Use Case</b>: High-throughput messaging, event sourcing, log aggregation.
+- <b>Scalability</b>: High with partitioned topics.
+- <b>Delivery Guarantees</b>: Configurable (at-least-once, exactly-once).
+- <b>Integration</b>: Broad ecosystem with various connectors.
+- <b>Delivery Models</b>: Pull-based consumer groups.
+
+### Key Differences
+- <b>Management</b>: Pub/Sub and SQS are managed services, while Kafka is typically self-managed or via managed services like Confluent.
+- <b>Use Case Focus</b>: Pub/Sub and Kafka are ideal for real-time processing, whereas SQS is great for decoupling microservices and handling asynchronous tasks.
+- <b>Delivery Models</b>: Pub/Sub supports push and pull, SQS supports pull with long polling, and Kafka primarily uses pull with consumer groups.
+- <b>Scalability</b>: All three are highly scalable, but Kafka offers the most control over performance tuning.
+- <b>Integration</b>: Pub/Sub integrates well with Google Cloud, SQS with AWS, and Kafka has a broad integration ecosystem.
 
 ## Installation
 
-Please make sure to initialize a Go module before installing common-go/sqs:
+Please make sure to initialize a Go module before installing core-go/sqs:
 
 ```shell
-go get -u github.com/common-go/sqs
+go get -u github.com/core-go/sqs
 ```
 
 Import:
 
 ```go
-import "github.com/common-go/sqs"
+import "github.com/core-go/sqs"
 ```

health_checker.go

@@ -38,6 +38,7 @@ func (h *HealthChecker) Name() string {
 
 func (h *HealthChecker) Check(ctx context.Context) (map[string]interface{}, error) {
 	res := make(map[string]interface{})
+	h.Client.Config.HTTPClient.Timeout = h.Timeout
 	_, err := h.Client.GetQueueUrl(&sqs.GetQueueUrlInput{
 		QueueName: h.QueueName,
 	})
@@ -48,6 +49,9 @@ func (h *HealthChecker) Build(ctx context.Context, data map[string]interface{},
 	if err == nil {
 		return data
 	}
+	if data == nil {
+		data = make(map[string]interface{}, 0)
+	}
 	data["error"] = err.Error()
 	return data
 }

queue_sender.go

@@ -0,0 +1,44 @@
+package sqs
+
+import (
+	"context"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/sqs"
+)
+
+type QueueSender struct {
+	Client       *sqs.SQS
+	DelaySeconds *int64 //could be 10
+}
+
+func NewQueueSender(client *sqs.SQS, delaySeconds int64) *QueueSender {
+	return &QueueSender{Client: client, DelaySeconds: &delaySeconds}
+}
+func (p *QueueSender) Send(ctx context.Context, queueName string, data []byte, attributes map[string]string) error {
+	queueUrl, er0 := GetQueueUrl(p.Client, queueName)
+	if er0 != nil {
+		return er0
+	}
+	attrs := MapToAttributes(attributes)
+	s := string(data)
+	_, err := p.Client.SendMessage(&sqs.SendMessageInput{
+		DelaySeconds:      p.DelaySeconds,
+		MessageAttributes: attrs,
+		MessageBody:       aws.String(s),
+		QueueUrl:          &queueUrl,
+	})
+	return err
+}
+func (p *QueueSender) SendBody(ctx context.Context, queueName string, data []byte) error {
+	queueUrl, er0 := GetQueueUrl(p.Client, queueName)
+	if er0 != nil {
+		return er0
+	}
+	s := string(data)
+	_, err := p.Client.SendMessage(&sqs.SendMessageInput{
+		DelaySeconds: p.DelaySeconds,
+		MessageBody:  aws.String(s),
+		QueueUrl:     &queueUrl,
+	})
+	return err
+}

receiver.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sqs"
-	"github.com/common-go/mq"
 )
 
 type Receiver struct {
@@ -13,6 +12,7 @@ type Receiver struct {
 	AckOnConsume      bool
 	VisibilityTimeout int64 // should be 20 (seconds)
 	WaitTimeSeconds   int64 // should be 0
+	LogError          func(ctx context.Context, msg string)
 }
 
 func NewReceiverByQueueName(client *sqs.SQS, queueName string, ackOnConsume bool, visibilityTimeout int64, waitTimeSeconds int64) (*Receiver, error) {
@@ -27,53 +27,123 @@ func NewReceiver(client *sqs.SQS, queueURL string, ackOnConsume bool, visibility
 	return &Receiver{Client: client, QueueURL: &queueURL, AckOnConsume: ackOnConsume, VisibilityTimeout: visibilityTimeout, WaitTimeSeconds: waitTimeSeconds}
 }
 
-func (c *Receiver) Receive(ctx context.Context, handle func(context.Context, *mq.Message, error) error) {
+func (c *Receiver) Receive(ctx context.Context, handle func(context.Context, []byte, map[string]string)) {
 	var result *sqs.ReceiveMessageOutput
 	var er1 error
-	loop:
-		result, er1 = c.Client.ReceiveMessage(&sqs.ReceiveMessageInput{
-			AttributeNames: []*string{
-				aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
-			},
-			MessageAttributeNames: []*string{
-				aws.String(sqs.QueueAttributeNameAll),
-			},
-			QueueUrl:            c.QueueURL,
-			MaxNumberOfMessages: aws.Int64(1),
-			VisibilityTimeout:   aws.Int64(c.VisibilityTimeout), // 20 seconds
-			WaitTimeSeconds:     aws.Int64(c.WaitTimeSeconds),
-		})
-		if er1 != nil {
-			handle(ctx, nil, er1)
-		} else {
-			if len(result.Messages) > 0 {
-				m := result.Messages[0]
-				data := []byte(*m.Body)
-				attributes := PtrToMap(m.Attributes)
-				message := mq.Message{
-					Id:         *m.MessageId,
-					Data:       data,
-					Attributes: attributes,
-					Raw:        m,
+loop:
+	result, er1 = c.Client.ReceiveMessage(&sqs.ReceiveMessageInput{
+		AttributeNames: []*string{
+			aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
+		},
+		MessageAttributeNames: []*string{
+			aws.String(sqs.QueueAttributeNameAll),
+		},
+		QueueUrl:            c.QueueURL,
+		MaxNumberOfMessages: aws.Int64(1),
+		VisibilityTimeout:   aws.Int64(c.VisibilityTimeout), // 20 seconds
+		WaitTimeSeconds:     aws.Int64(c.WaitTimeSeconds),
+	})
+	if er1 != nil {
+		c.LogError(ctx, "Error when subscribe: "+er1.Error())
+	} else {
+		if len(result.Messages) > 0 {
+			m := result.Messages[0]
+			data := []byte(*m.Body)
+			attributes := PtrToMap(m.Attributes)
+			if c.AckOnConsume {
+				_, er2 := c.Client.DeleteMessage(&sqs.DeleteMessageInput{
+					QueueUrl:      c.QueueURL,
+					ReceiptHandle: result.Messages[0].ReceiptHandle,
+				})
+				if er2 != nil {
+					c.LogError(ctx, "Error when delete message: "+er2.Error())
+				} else {
+					handle(ctx, data, attributes)
 				}
-				if c.AckOnConsume {
-					_, er2 := c.Client.DeleteMessage(&sqs.DeleteMessageInput{
-						QueueUrl:      c.QueueURL,
-						ReceiptHandle: result.Messages[0].ReceiptHandle,
-					})
-					if er2 != nil {
-						handle(ctx, nil, er2)
-					} else {
-						handle(ctx, &message, nil)
-					}
+			} else {
+				handle(ctx, data, attributes)
+			}
+		}
+	}
+	goto loop
+}
+func (c *Receiver) ReceiveBody(ctx context.Context, handle func(context.Context, []byte)) {
+	var result *sqs.ReceiveMessageOutput
+	var er1 error
+loop:
+	result, er1 = c.Client.ReceiveMessage(&sqs.ReceiveMessageInput{
+		AttributeNames: []*string{
+			aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
+		},
+		MessageAttributeNames: []*string{
+			aws.String(sqs.QueueAttributeNameAll),
+		},
+		QueueUrl:            c.QueueURL,
+		MaxNumberOfMessages: aws.Int64(1),
+		VisibilityTimeout:   aws.Int64(c.VisibilityTimeout), // 20 seconds
+		WaitTimeSeconds:     aws.Int64(c.WaitTimeSeconds),
+	})
+	if er1 != nil {
+		c.LogError(ctx, "Error when subscribe: "+er1.Error())
+	} else {
+		if len(result.Messages) > 0 {
+			m := result.Messages[0]
+			data := []byte(*m.Body)
+			if c.AckOnConsume {
+				_, er2 := c.Client.DeleteMessage(&sqs.DeleteMessageInput{
+					QueueUrl:      c.QueueURL,
+					ReceiptHandle: result.Messages[0].ReceiptHandle,
+				})
+				if er2 != nil {
+					c.LogError(ctx, "Error when delete message: "+er2.Error())
 				} else {
-					handle(ctx, &message, nil)
+					handle(ctx, data)
 				}
+			} else {
+				handle(ctx, data)
 			}
 		}
-		goto loop
+	}
+	goto loop
+}
+func (c *Receiver) ReceiveMessage(ctx context.Context, handle func(context.Context, *sqs.Message)) {
+	var result *sqs.ReceiveMessageOutput
+	var er1 error
+loop:
+	result, er1 = c.Client.ReceiveMessage(&sqs.ReceiveMessageInput{
+		AttributeNames: []*string{
+			aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
+		},
+		MessageAttributeNames: []*string{
+			aws.String(sqs.QueueAttributeNameAll),
+		},
+		QueueUrl:            c.QueueURL,
+		MaxNumberOfMessages: aws.Int64(1),
+		VisibilityTimeout:   aws.Int64(c.VisibilityTimeout), // 20 seconds
+		WaitTimeSeconds:     aws.Int64(c.WaitTimeSeconds),
+	})
+	if er1 != nil {
+		c.LogError(ctx, "Error when subscribe: "+er1.Error())
+	} else {
+		if len(result.Messages) > 0 {
+			m := result.Messages[0]
+			if c.AckOnConsume {
+				_, er2 := c.Client.DeleteMessage(&sqs.DeleteMessageInput{
+					QueueUrl:      c.QueueURL,
+					ReceiptHandle: result.Messages[0].ReceiptHandle,
+				})
+				if er2 != nil {
+					c.LogError(ctx, "Error when delete message: "+er2.Error())
+				} else {
+					handle(ctx, m)
+				}
+			} else {
+				handle(ctx, m)
+			}
+		}
+	}
+	goto loop
 }
-
 func PtrToMap(m map[string]*string) map[string]string {
 	attributes := make(map[string]string)
 	for k, v := range m {

sender.go

@@ -12,27 +12,42 @@ type Sender struct {
 	DelaySeconds *int64 //could be 10
 }
 
-func NewSenderByQueueName(client *sqs.SQS, queueName string, delaySeconds *int64) (*Sender, error) {
+func NewSenderByQueueName(client *sqs.SQS, queueName string, delaySeconds int64) (*Sender, error) {
 	queueUrl, err := GetQueueUrl(client, queueName)
 	if err != nil {
 		return nil, err
 	}
 	return NewSender(client, queueUrl, delaySeconds), nil
 }
 
-func NewSender(client *sqs.SQS, queueURL string, delaySeconds *int64) *Sender {
-	return &Sender{Client: client, QueueURL: &queueURL, DelaySeconds: delaySeconds}
+func NewSender(client *sqs.SQS, queueURL string, delaySeconds int64) *Sender {
+	return &Sender{Client: client, QueueURL: &queueURL, DelaySeconds: &delaySeconds}
 }
-
-func (p *Sender) Send(ctx context.Context, data []byte, attributes map[string]string) (string, error) {
+func (p *Sender) Send(ctx context.Context, data []byte, attributes map[string]string) error {
 	attrs := MapToAttributes(attributes)
 	s := string(data)
-	result, err := p.Client.SendMessage(&sqs.SendMessageInput{
+	_, err := p.Client.SendMessage(&sqs.SendMessageInput{
 		DelaySeconds:      p.DelaySeconds,
 		MessageAttributes: attrs,
 		MessageBody:       aws.String(s),
 		QueueUrl:          p.QueueURL,
 	})
+	return err
+}
+func (p *Sender) SendBody(ctx context.Context, data []byte) error {
+	s := string(data)
+	_, err := p.Client.SendMessage(&sqs.SendMessageInput{
+		DelaySeconds: p.DelaySeconds,
+		MessageBody:  aws.String(s),
+		QueueUrl:     p.QueueURL,
+	})
+	return err
+}
+func (p *Sender) SendMessage(msg *sqs.SendMessageInput) (string, error) {
+	if msg == nil {
+		return "", nil
+	}
+	result, err := p.Client.SendMessage(msg)
 	if result != nil && result.MessageId != nil {
 		return *result.MessageId, err
 	} else {

sqs.go

@@ -9,10 +9,10 @@ import (
 
 type (
 	Config struct {
-		Region          string `mapstructure:"region" json:"region,omitempty" gorm:"column:region" bson:"region,omitempty" dynamodbav:"region,omitempty" firestore:"region,omitempty"`
-		AccessKeyID     string `mapstructure:"access_key_id" json:"accessKeyID,omitempty" gorm:"column:accessKeyID" bson:"accessKeyID,omitempty" dynamodbav:"accessKeyID,omitempty" firestore:"accessKeyID,omitempty"`
-		SecretAccessKey string `mapstructure:"secret_access_key" json:"secretAccessKey,omitempty" gorm:"column:secretaccesskey" bson:"secretAccessKey,omitempty" dynamodbav:"secretAccessKey,omitempty" firestore:"secretAccessKey,omitempty"`
-		QueueName       string `mapstructure:"queue_name" json:"queueName,omitempty" gorm:"column:token" bson:"queueName,omitempty" dynamodbav:"queueName,omitempty" firestore:"queueName,omitempty"`
+		Region          string `yaml:"region" mapstructure:"region" json:"region,omitempty" gorm:"column:region" bson:"region,omitempty" dynamodbav:"region,omitempty" firestore:"region,omitempty"`
+		AccessKeyID     string `yaml:"access_key_id" mapstructure:"access_key_id" json:"accessKeyID,omitempty" gorm:"column:accessKeyID" bson:"accessKeyID,omitempty" dynamodbav:"accessKeyID,omitempty" firestore:"accessKeyID,omitempty"`
+		SecretAccessKey string `yaml:"secret_access_key" mapstructure:"secret_access_key" json:"secretAccessKey,omitempty" gorm:"column:secretaccesskey" bson:"secretAccessKey,omitempty" dynamodbav:"secretAccessKey,omitempty" firestore:"secretAccessKey,omitempty"`
+		QueueName       string `yaml:"a" mapstructure:"queue_name" json:"queueName,omitempty" gorm:"column:token" bson:"queueName,omitempty" dynamodbav:"queueName,omitempty" firestore:"queueName,omitempty"`
 	}
 )
 
@@ -35,4 +35,4 @@ func Connect(config Config) (*sqs.SQS, error) {
 
 func ConnectWithSession(session *session.Session) *sqs.SQS {
 	return sqs.New(session)
-}
+}