Skip to content

Commit 19d2891

Browse files
author
Gabriel Cataldo
committed
Finish documentation and README
1 parent bb0f821 commit 19d2891

15 files changed

+834
-114
lines changed

‎README.md

+399-1
Large diffs are not rendered by default.

‎_example/consumer/main.go

+108
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package main
2+
3+
import (
4+
"github.com/GabrielHCataldo/go-aws-sqs/sqs"
5+
"github.com/GabrielHCataldo/go-aws-sqs/sqs/option"
6+
"github.com/GabrielHCataldo/go-logger/logger"
7+
"os"
8+
"os/signal"
9+
"time"
10+
)
11+
12+
type test struct {
13+
Name string `json:"name,omitempty"`
14+
BirthDate time.Time `json:"birthDate,omitempty"`
15+
Emails []string `json:"emails,omitempty"`
16+
Bank bank `json:"bank,omitempty"`
17+
Map map[string]any
18+
}
19+
20+
type bank struct {
21+
Account string `json:"account,omitempty"`
22+
Digits string `json:"digits,omitempty"`
23+
Balance float64 `json:"balance,omitempty"`
24+
}
25+
26+
type messageAttTest struct {
27+
Name string `json:"account,omitempty"`
28+
Text string `json:"text,omitempty"`
29+
Balance float64 `json:"balance,omitempty"`
30+
Bool bool `json:"bool"`
31+
Int int `json:"int"`
32+
SubStruct test `json:"subStruct,omitempty"`
33+
PointerTest *test `json:"pointerBank,omitempty"`
34+
Map map[string]any
35+
Any any
36+
EmptyString string `json:"emptyString,omitempty"`
37+
HideString string `json:"-"`
38+
}
39+
40+
func main() {
41+
simpleReceiveMessage()
42+
simpleReceiveMessageStruct()
43+
receiveMessage()
44+
completeOptions()
45+
simpleReceiveMessageAsync()
46+
}
47+
48+
func simpleReceiveMessage() {
49+
sqs.SimpleReceiveMessage(os.Getenv("SQS_QUEUE_TEST_STRING_URL"), handlerSimple)
50+
}
51+
52+
func simpleReceiveMessageStruct() {
53+
sqs.SimpleReceiveMessage(os.Getenv("SQS_QUEUE_TEST_URL"), handler)
54+
}
55+
56+
func receiveMessage() {
57+
sqs.ReceiveMessage(os.Getenv("SQS_QUEUE_TEST_STRING_URL"), handlerReceiveMessage)
58+
}
59+
60+
func completeOptions() {
61+
opt := option.NewConsumer().
62+
// HTTP communication customization options with AWS SQS
63+
SetHttpClient(option.HttpClient{}).
64+
// print logs (default: false)
65+
SetDebugMode(true).
66+
// If true remove the message from the queue after successfully processed (handler error return is null) (default: false)
67+
SetDeleteMessageProcessedSuccess(true).
68+
// Duration time to process the message, timeout applied in the past context. (default: 5 seconds)
69+
SetConsumerMessageTimeout(5 * time.Second).
70+
// Delay to run the next search for messages in the queue (default: 0)
71+
SetDelayQueryLoop(5 * time.Second).
72+
// The maximum number of messages to return. 1 a 10 (default: 10)
73+
SetMaxNumberOfMessages(10).
74+
// The maximum number of messages to return. 1 a 10 (default: 0)
75+
SetVisibilityTimeout(5 * time.Second).
76+
// The duration that the received messages are hidden from subsequent
77+
// retrieve requests after being retrieved by a ReceiveMessage request.
78+
SetReceiveRequestAttemptId("").
79+
// The duration for which the call waits for a message to arrive in
80+
// the queue before returning (default: 0)
81+
SetWaitTimeSeconds(1 * time.Second)
82+
sqs.SimpleReceiveMessage(os.Getenv("SQS_QUEUE_TEST_URL"), handler, opt)
83+
}
84+
85+
func simpleReceiveMessageAsync() {
86+
sqs.SimpleReceiveMessageAsync(os.Getenv("SQS_QUEUE_TEST_URL"), handler, option.NewConsumer().SetDebugMode(true))
87+
c := make(chan os.Signal, 1)
88+
signal.Notify(c, os.Interrupt)
89+
select {
90+
case <-c:
91+
logger.Info("Stopped application!")
92+
}
93+
}
94+
95+
func handler(ctx *sqs.SimpleContext[test]) error {
96+
logger.Debug("ctx simple body struct to process message:", ctx)
97+
return nil
98+
}
99+
100+
func handlerSimple(ctx *sqs.SimpleContext[string]) error {
101+
logger.Debug("ctx simple to process message:", ctx)
102+
return nil
103+
}
104+
105+
func handlerReceiveMessage(ctx *sqs.Context[string, messageAttTest]) error {
106+
logger.Debug("ctx to process message:", ctx)
107+
return nil
108+
}

‎_example/producer/main.go

+150
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"github.com/GabrielHCataldo/go-aws-sqs/sqs"
6+
"github.com/GabrielHCataldo/go-aws-sqs/sqs/option"
7+
"github.com/GabrielHCataldo/go-logger/logger"
8+
"os"
9+
"time"
10+
)
11+
12+
type test struct {
13+
Name string `json:"name,omitempty"`
14+
BirthDate time.Time `json:"birthDate,omitempty"`
15+
Emails []string `json:"emails,omitempty"`
16+
Bank bank `json:"bank,omitempty"`
17+
Map map[string]any
18+
}
19+
20+
type bank struct {
21+
Account string `json:"account,omitempty"`
22+
Digits string `json:"digits,omitempty"`
23+
Balance float64 `json:"balance,omitempty"`
24+
}
25+
26+
type messageAttTest struct {
27+
Name string `json:"account,omitempty"`
28+
Text string `json:"text,omitempty"`
29+
Balance float64 `json:"balance,omitempty"`
30+
Bool bool `json:"bool"`
31+
Int int `json:"int"`
32+
SubStruct test `json:"subStruct,omitempty"`
33+
PointerTest *test `json:"pointerBank,omitempty"`
34+
Map map[string]any
35+
Any any
36+
EmptyString string `json:"emptyString,omitempty"`
37+
HideString string `json:"-"`
38+
}
39+
40+
func main() {
41+
simple()
42+
simpleAsync()
43+
structBody()
44+
mapBody()
45+
completeOptions()
46+
}
47+
48+
func simple() {
49+
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
50+
defer cancel()
51+
body := "body test"
52+
message, err := sqs.SendMessage(ctx, os.Getenv("SQS_QUEUE_TEST_STRING_URL"), body)
53+
if err != nil {
54+
logger.Error("error send message:", err)
55+
} else {
56+
logger.Info("message sent successfully:", message)
57+
}
58+
}
59+
60+
func simpleAsync() {
61+
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
62+
defer cancel()
63+
body := "body test"
64+
sqs.SendMessageAsync(ctx, os.Getenv("SQS_QUEUE_TEST_STRING_URL"), body)
65+
}
66+
67+
func structBody() {
68+
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
69+
defer cancel()
70+
body := initTestStruct()
71+
message, err := sqs.SendMessage(ctx, os.Getenv("SQS_QUEUE_TEST_URL"), body)
72+
if err != nil {
73+
logger.Error("error send message:", err)
74+
} else {
75+
logger.Info("message sent successfully:", message)
76+
}
77+
}
78+
79+
func mapBody() {
80+
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
81+
defer cancel()
82+
body := initTestMap()
83+
message, err := sqs.SendMessage(ctx, os.Getenv("SQS_QUEUE_TEST_STRING_URL"), body)
84+
if err != nil {
85+
logger.Error("error send message:", err)
86+
} else {
87+
logger.Info("message sent successfully:", message)
88+
}
89+
}
90+
91+
func completeOptions() {
92+
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
93+
defer cancel()
94+
body := initTestStruct()
95+
opt := option.NewProducer().
96+
// HTTP communication customization options with AWS SQS
97+
SetHttpClient(option.HttpClient{}).
98+
// print logs (default: false)
99+
SetDebugMode(true).
100+
// delay to delay the availability of message processing (default: 0)
101+
SetDelaySeconds(5 * time.Second).
102+
// Message attributes, must be of type Map or Struct, other types are not acceptable.
103+
SetMessageAttributes(initMessageAttTest()).
104+
// The message system attribute to send
105+
SetMessageSystemAttributes(option.MessageSystemAttributes{}).
106+
// This parameter applies only to FIFO (first-in-first-out) queues. The token used for deduplication of sent messages.
107+
SetMessageDeduplicationId("").
108+
// This parameter applies only to FIFO (first-in-first-out) queues. The tag that specifies that a message belongs to a specific message group.
109+
SetMessageGroupId("")
110+
message, err := sqs.SendMessage(ctx, os.Getenv("SQS_QUEUE_TEST_URL"), body, opt)
111+
if err != nil {
112+
logger.Error("error send message:", err)
113+
} else {
114+
logger.Info("message sent successfully:", message)
115+
}
116+
}
117+
118+
func initTestStruct() test {
119+
b := bank{
120+
Account: "123456",
121+
Digits: "2",
122+
Balance: 200.12,
123+
}
124+
return test{
125+
Name: "Test Name",
126+
BirthDate: time.Now(),
127+
Emails: []string{"test@gmail.com", "gabriel@gmail.com", "gabriel.test@gmail.com"},
128+
Bank: b,
129+
Map: map[string]any{"int": 1, "bool": true, "float": 1.23, "string": "text test"},
130+
}
131+
}
132+
133+
func initTestMap() map[string]any {
134+
return map[string]any{"int": 1, "bool": true, "float": 1.23, "string": "text test"}
135+
}
136+
137+
func initMessageAttTest() messageAttTest {
138+
t := initTestStruct()
139+
return messageAttTest{
140+
Name: "Name test producer",
141+
Text: "Text field",
142+
Balance: 10.23,
143+
Bool: true,
144+
Int: 3,
145+
SubStruct: t,
146+
PointerTest: &t,
147+
Map: initTestMap(),
148+
HideString: "hide test",
149+
}
150+
}

‎go.mod

+2-2
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
1-
module go-aws-sqs
1+
module github.com/GabrielHCataldo/go-aws-sqs
22

3-
go 1.21
3+
go 1.21.3
44

55
require (
66
github.com/GabrielHCataldo/go-logger v1.0.8

‎gopher-sqs.png

29.9 KB
Loading

‎sqs/consumer.go

+59-18
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,11 @@ package sqs
33
import (
44
"context"
55
"fmt"
6+
"github.com/GabrielHCataldo/go-aws-sqs/internal/client"
7+
"github.com/GabrielHCataldo/go-aws-sqs/internal/util"
8+
"github.com/GabrielHCataldo/go-aws-sqs/sqs/option"
69
"github.com/aws/aws-sdk-go-v2/service/sqs"
710
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
8-
"go-aws-sqs/internal/client"
9-
"go-aws-sqs/internal/util"
10-
"go-aws-sqs/sqs/option"
1111
"reflect"
1212
"time"
1313
)
@@ -89,6 +89,11 @@ type HandlerConsumerFunc[Body, MessageAttributes any] func(ctx *Context[Body, Me
8989
// HandlerSimpleConsumerFunc is a function that consumes a message and returns an error if a failure occurs while processing the message.
9090
type HandlerSimpleConsumerFunc[Body any] func(ctx *SimpleContext[Body]) error
9191

92+
type channelMessageProcessed struct {
93+
Err error
94+
Signal *chan struct{}
95+
}
96+
9297
var ctxInterrupt context.Context
9398

9499
// ReceiveMessage Works as a repeating job, when triggered, it will fetch messages from the indicated queue
@@ -238,6 +243,7 @@ func receiveMessage[Body, MessageAttributes any](
238243
}
239244
loggerInfo(opt.DebugMode, "Start process received messages size:", len(output.Messages))
240245
processMessages[Body, MessageAttributes](queueUrl, output, handler, opt)
246+
time.Sleep(opt.DelayQueryLoop)
241247
}
242248
}
243249

@@ -275,26 +281,61 @@ func processMessages[Body, MessageAttributes any](
275281
handler HandlerConsumerFunc[Body, MessageAttributes],
276282
opt option.Consumer,
277283
) {
278-
ctx, cancel := context.WithTimeout(context.TODO(), opt.ConsumerMessageTimeout)
279-
defer cancel()
280284
var count int
281285
var mgsS, mgsF []string
282286
for _, message := range output.Messages {
283-
nCtx, err := prepareContextConsumer[Body, MessageAttributes](ctx, queueUrl, message)
284-
if err != nil {
285-
loggerErr(opt.DebugMode, "error prepare context to consumer:", err)
286-
return
287-
}
288-
err = handler(nCtx)
289-
appendMessagesByResult(nCtx.Message.Id, err, mgsS, mgsF)
290-
if opt.DeleteMessageProcessedSuccess {
291-
go deleteMessage(queueUrl, *message.ReceiptHandle)
292-
}
287+
mgsS, mgsF = processMessage(queueUrl, handler, message, opt)
293288
count++
294289
}
295290
loggerInfo(opt.DebugMode, "Finish process messages!", "processed:", count, "success:", mgsS, "failed:", mgsF)
296291
}
297292

293+
func processMessage[Body, MessageAttributes any](
294+
queueUrl string,
295+
handler HandlerConsumerFunc[Body, MessageAttributes],
296+
message types.Message,
297+
opt option.Consumer,
298+
) (mgsS, mgsF []string) {
299+
ctx, cancel := context.WithTimeout(context.TODO(), opt.ConsumerMessageTimeout)
300+
defer cancel()
301+
ctxConsumer, err := prepareContextConsumer[Body, MessageAttributes](ctx, queueUrl, message)
302+
if err != nil {
303+
loggerErr(opt.DebugMode, "error prepare context to consumer:", err)
304+
return
305+
}
306+
signal := make(chan struct{}, 1)
307+
channel := channelMessageProcessed{
308+
Signal: &signal,
309+
}
310+
go processHandler(ctxConsumer, handler, opt, &channel)
311+
select {
312+
case <-ctx.Done():
313+
appendMessagesByResult(ctxConsumer.Message.Id, ctx.Err(), &mgsS, &mgsF)
314+
break
315+
case <-*channel.Signal:
316+
appendMessagesByResult(*message.MessageId, channel.Err, &mgsS, &mgsF)
317+
break
318+
}
319+
return mgsS, mgsF
320+
}
321+
322+
func processHandler[Body, MessageAttributes any](
323+
ctx *Context[Body, MessageAttributes],
324+
handler HandlerConsumerFunc[Body, MessageAttributes],
325+
opt option.Consumer,
326+
channel *channelMessageProcessed,
327+
) {
328+
err := handler(ctx)
329+
if ctx.Err() != nil {
330+
return
331+
}
332+
if err == nil && opt.DeleteMessageProcessedSuccess {
333+
go deleteMessage(ctx.QueueUrl, ctx.Message.ReceiptHandle)
334+
}
335+
channel.Err = err
336+
*channel.Signal <- struct{}{}
337+
}
338+
298339
func prepareContextConsumer[Body, MessageAttributes any](
299340
ctx context.Context,
300341
queueUrl string,
@@ -334,11 +375,11 @@ func prepareContextConsumer[Body, MessageAttributes any](
334375
return ctxConsumer, nil
335376
}
336377

337-
func appendMessagesByResult(messageId string, err error, mgsS, mgsF []string) {
378+
func appendMessagesByResult(messageId string, err error, mgsS, mgsF *[]string) {
338379
if err != nil {
339-
mgsF = append(mgsF, messageId)
380+
*mgsF = append(*mgsF, messageId)
340381
} else {
341-
mgsS = append(mgsS, messageId)
382+
*mgsS = append(*mgsS, messageId)
342383
}
343384
}
344385

0 commit comments

Comments
 (0)