aws
/
sns-sqs-test
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

instead of ack+message-id just -ack message-id

master
Bel LaPointe 2023-02-06 08:10:29 -07:00
parent 839289ea81
commit 4beaa4c387
1 changed files with 11 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
main.go
View File

@ -23,16 +23,15 @@ import (
) )
type Config struct { type Config struct {
AWSA string AWSA string
AWSU string AWSU string
AWSP string AWSP string
AWSR string AWSR string
Topic string Topic string
Message string Message string
REMSWrap bool REMSWrap bool
MessageID string Ack string
Ack bool Pub bool
Pub bool
} }
type isoc struct{} type isoc struct{}
@ -84,9 +83,8 @@ func getConfig() (Config, error) {
flag.StringVar(&c.AWSR, "awsr", "eu-central-1", "aws region") flag.StringVar(&c.AWSR, "awsr", "eu-central-1", "aws region")
flag.StringVar(&c.Topic, "topic", "rems-queue-prod-fra1-priority-low", "sns topic/sqs queue") flag.StringVar(&c.Topic, "topic", "rems-queue-prod-fra1-priority-low", "sns topic/sqs queue")
flag.StringVar(&c.Message, "message", `{"messageType": "NoOpMessage"}`, "published message") flag.StringVar(&c.Message, "message", `{"messageType": "NoOpMessage"}`, "published message")
flag.StringVar(&c.MessageID, "message-id", "", "sqs message id to ack") flag.StringVar(&c.Ack, "ack", "", "sqs message id to ack")
flag.BoolVar(&c.REMSWrap, "rems-wrap", true, "rems wrap message") flag.BoolVar(&c.REMSWrap, "rems-wrap", true, "rems wrap message")
flag.BoolVar(&c.Ack, "ack", true, "ack message matching message-id")
flag.BoolVar(&c.Pub, "pub", false, "publish message") flag.BoolVar(&c.Pub, "pub", false, "publish message")
flag.Parse() flag.Parse()
@ -148,7 +146,7 @@ func (c Config) Consume(ctx context.Context) error {
wrap.Unwrap([]byte(*message.Body), &v, isoc{}) wrap.Unwrap([]byte(*message.Body), &v, isoc{})
} }
log.Printf("message in queue: %s: %s: %+v", *message.MessageId, *message.Body, v) log.Printf("message in queue: %s: %s: %+v", *message.MessageId, *message.Body, v)
if *message.MessageId != c.MessageID { if *message.MessageId != c.Ack {
if err := c.unack(ctx, message); err != nil { if err := c.unack(ctx, message); err != nil {
return err return err
} }
@ -175,9 +173,6 @@ func (c Config) unack(ctx context.Context, message *sqs.Message) error {
} }
func (c Config) ack(ctx context.Context, message *sqs.Message) error { func (c Config) ack(ctx context.Context, message *sqs.Message) error {
if !c.Ack {
return c.unack(ctx, message)
}
client := sqs.New(session.Must(session.NewSession()), c.awsc()) client := sqs.New(session.Must(session.NewSession()), c.awsc())
if _, err := client.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{ if _, err := client.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{
QueueUrl: aws.String(c.sqsurl()), QueueUrl: aws.String(c.sqsurl()),