-
Notifications
You must be signed in to change notification settings - Fork 1
/
consumer.go
91 lines (77 loc) · 2.29 KB
/
consumer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
package consumer
import (
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/sqs"
"log"
"os"
"os/signal"
"syscall"
)
//Consumer holds the consumer data
type Consumer struct {
queueURL string
messagesChannel chan []*sqs.Message
handler func(m *sqs.Message) error
config *Config
receiver SqsReceiver
}
//Config holds the configuration for consuming and processing the queue
type Config struct {
AwsSession *session.Session
SqsMaxNumberOfMessages int64
SqsMessageVisibilityTimeout int64
Receivers int
PollDelayInMilliseconds int
}
var sess *session.Session
//New creates a new Queue consumer
func New(queueURL string, handler func(m *sqs.Message) error, config *Config) Consumer {
sess = config.AwsSession
c := make(chan []*sqs.Message)
shutdown := make(chan os.Signal, 1)
signal.Notify(shutdown, syscall.SIGINT, syscall.SIGTERM)
r := SqsReceiver{
queueURL: queueURL,
messagesChannel: c,
shutdown: shutdown,
sess: sess,
visibilityTimeout: config.SqsMessageVisibilityTimeout,
maxNumberOfMessages: config.SqsMaxNumberOfMessages,
pollDelayInMilliseconds: config.PollDelayInMilliseconds,
}
return Consumer{
queueURL: queueURL,
messagesChannel: c,
handler: handler,
config: config,
receiver: r,
}
}
//Start initiates the queue consumption process
func (c *Consumer) Start() {
log.Println("Starting to consume", c.queueURL)
c.startReceivers()
c.startProcessor()
}
// startReceivers starts N (defined in NumberOfMessageReceivers) goroutines to poll messages from SQS
func (c *Consumer) startReceivers() {
for i := 0; i < c.config.Receivers; i++ {
go c.receiver.receiveMessages()
}
}
// startProcessor starts a goroutine to handle each message from messagesChannel
func (c *Consumer) startProcessor() {
queue := sqs.New(sess)
p := Processor{
queueURL: c.queueURL,
queue: queue,
handler: c.handler,
}
for messages := range c.messagesChannel {
go p.processMessages(messages)
}
}
//SetPollDelay increases time between message poll
func (c *Consumer) SetPollDelay(delayBetweenPoolsInMilliseconds int) {
c.receiver.pollDelayInMilliseconds = delayBetweenPoolsInMilliseconds
}