[WIP] v1.0 preparations #17

roblaszczak · 2024-08-20T19:40:52Z

No description provided.

roblaszczak · 2024-08-23T16:22:47Z

README.md

@@ -16,8 +16,7 @@ You can run the unit tests by simply running  the command: `go test -cover -race

 #### Development and Testing

-To try the  in test mode (using [goaws](https://github.com/p4tin/goaws))
+To try the  in test mode (using [localstack](https://hub.docker.com/r/localstack/localstack))


goaws is not compatible with newer AWS SDK - I needed to replace it

I don't know why but I remember experiencing strange behaviours with localstack.

unfortunetly goaws doesn't work with never AWS SDK: Admiral-Piett/goaws#279

but it looks lighter than localstack so we can potentially go back

roblaszczak · 2024-08-23T16:24:29Z

sns/pub_test.go

@@ -93,7 +93,7 @@ func createPubSub(t *testing.T) (message.Publisher, message.Subscriber) {

 	sub, err := sqs.NewSubscriber(sqs.SubscriberConfig{
 		AWSConfig: cfg,
-		CreateQueueInitializerConfig: sqs.QueueConfigAtrributes{


there was a typo in QueueConfigAttributes

roblaszczak · 2024-08-23T16:25:40Z

sqs/config.go

+	CreateQueueInitializerConfig QueueConfigAttributes
+
+	GenerateCreateQueueInput GenerateCreateQueueInputFunc
+
+	GenerateGetQueueUrlInput GenerateGetQueueUrlInputFunc
+
+	GenerateReceiveMessageInput GenerateReceiveMessageInputFunc
+
+	GenerateDeleteMessageInput GenerateDeleteMessageInputFunc


I've added config for all things where we can specify in AWS requests - all those has sane defaults

looking for feedback

I need to add comments here

roblaszczak · 2024-08-23T16:26:35Z

sqs/config.go

+	return &sqs.ReceiveMessageInput{
+		QueueUrl:              aws.String(queueURL),
+		MessageAttributeNames: []string{"All"},
+		WaitTimeSeconds:       20, // 20 is max at the moment


this is time that for what request send by sdk will wait for messages if they are none yet - to avoid calling AWS SDK like crazy

roblaszczak · 2024-08-23T16:27:32Z

sqs/marshaler.go

@@ -13,7 +13,7 @@ type Marshaler interface {
 	Marshal(msg *message.Message) (*types.Message, error)
 }

-type UnMarshaler interface {
+type Unmarshaler interface {


to make it consistent with stdlib and other watermill stuff

roblaszczak · 2024-08-23T16:29:26Z

sqs/sqs.go

-	b, _ := json.Marshal(q)
+// QueueConfigAttributesBool is a custom type for bool values in QueueConfigAttributes
+// that supports marshaling to string.
+type QueueConfigAttributesBool bool


it didn't worked earlier - it errored because json wasn't able to marshal bool to string - this type fixes it

roblaszczak · 2024-08-23T16:29:49Z

sqs/sqs.go

+	return []byte("false"), nil
+}
+
+func (q QueueConfigAttributes) Attributes() (map[string]string, error) {


i started to return errors from here as it was silently failing earlier 🙈

roblaszczak · 2024-08-23T16:30:35Z

sqs/sqs.go

-	getQueueOutput, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
-		QueueName: aws.String(topic),
-	})
+func getQueueUrl(ctx context.Context, sqsClient *sqs.Client, topic string, generateGetQueueUrlInput GenerateGetQueueUrlInputFunc) (*string, error) {


I made it private to not increase public API surface (and now it requires generateGetQueueUrlInput), but if anybody see it useful please let me know

roblaszczak · 2024-08-23T16:35:51Z

sqs/subscriber.go

@@ -94,33 +98,41 @@ func (s *Subscriber) receive(ctx context.Context, queueURL string, output chan *
 		"queue":    queueURL,
 	}

+	go func() {


without that tests were hanging - I added err check for all places that may return canceled ctx after that

roblaszczak · 2024-08-23T16:37:46Z

sqs/subscriber.go

 		}
+	case <-msg.Nacked():
+		// Do not delete message, it will be redelivered
+		return false // we don't want to process next messages to preserve order


we were not doing it earlier, but I guess that it's expected? if we'll continue we can process next messages and we may lose ordering

eventually we can add config here

roblaszczak · 2024-08-23T16:47:12Z

sqs/subscriber.go

+	return s.SubscribeInitializeWithContext(context.Background(), topic)
+}
+
+func (s *Subscriber) SubscribeInitializeWithContext(ctx context.Context, topic string) error {


just in case if someone wanted to pass ctx

m110 · 2024-08-26T09:13:15Z

sns/subscriber.go

+		SqsTopic:    sqsTopic,
+		SnsTopicArn: snsTopicArn,
+		SqsQueueArn: *sqsQueueArn,
+	})


Missing err check?

m110 · 2024-08-26T09:13:44Z

sns/subscriber.go

+
+	input, err := s.config.GenerateSubscribeInput(ctx, GenerateSubscribeInputParams{
+		SqsTopic:    sqsTopic,
+		SnsTopicArn: snsTopicArn,


Suggested change

SnsTopicArn: snsTopicArn,

SqsTopicArn: snsTopicArn,

Typo?

Mariscal6 · 2024-08-31T18:02:14Z

thanks @roblaszczak, looks much better

# Conflicts: # README.md # sqs/sqs.go

roblaszczak · 2024-09-05T19:55:24Z

sns/marshaler.go

@@ -8,18 +8,20 @@ import (
 	"github.com/ThreeDotsLabs/watermill/message"
 )

+// todo: check if it can be renamed


@Mariscal6 is UUID special in any way? or we could rename it to _watermill_uuid to avoid conflicts?

No it is not special, we can modify it

roblaszczak · 2024-09-05T19:56:06Z

sns/marshaler.go

 	// client side uuid
 	// there is a deduplication id that can be use for
 	// fifo queues
+	// todo: check how it works


@Mariscal6 connected to previous question 😉

Mariscal6 · 2024-09-14T08:25:52Z

cmd/sns-sqs/main.go

@@ -1,98 +0,0 @@
-package main


@roblaszczak I found these examples useful. Maybe we can put them in an example folder....

Mariscal6 · 2024-09-14T08:29:09Z

sns/marshaler.go

@@ -8,18 +8,20 @@ import (
 	"github.com/ThreeDotsLabs/watermill/message"
 )

+// todo: check if it can be renamed


No it is not special, we can modify it

Mariscal6 · 2024-09-14T08:35:15Z

sns/subscriber.go

+		SnsTopicArn: snsTopicArn,
+		SqsURL:      sqsURL,
+	})
+


Missing err check

Mariscal6 · 2024-09-14T08:36:18Z

sns/subscriber.go

+	return nil
+}
+
+func (s *Subscriber) setSqsQuePolicy(ctx context.Context, sqsQueueArn sqs.QueueArn, snsTopicArn TopicArn, sqsURL sqs.QueueURL) error {


setSqsQuePolicy to setSqsQueuePolicy?

roblaszczak added 6 commits August 20, 2024 21:37

v1.0 preparations

bc1730d

fix mapstructure

fdde1a5

go back to json

2fe21d5

really go back to json

18fdd82

fix WaitTimeSeconds

9f80d0e

fix tests and update emulator

0f9426d

roblaszczak commented Aug 23, 2024

View reviewed changes

roblaszczak added 8 commits August 24, 2024 10:23

cosmetics

a7bea07

more docs

dc70a8a

add support for sns subscribing + cleanups

e457144

use topic ARN as topic name for sns

e807740

rename

bd3a73f

added topic resolver

6713aa6

added queue url resolver

f6f0f37

cleanups

09f2b7c

m110 reviewed Aug 26, 2024

View reviewed changes

sns/subscriber.go

SqsTopic: sqsTopic,

SnsTopicArn: snsTopicArn,

SqsQueueArn: *sqsQueueArn,

})

Copy link

Member

m110 Aug 26, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing err check?

m110 reviewed Aug 26, 2024

View reviewed changes

downgrade watermill temporarly

5974b5f

fix locks and add support for setting queue access policy

67fc34d

roblaszczak added 3 commits September 5, 2024 21:47

cleanups

af4e21e

Merge branch 'refs/heads/master' into release

9a4cfe8

# Conflicts: # README.md # sqs/sqs.go

added DEVELOPMENT.md

1cb3844

roblaszczak marked this pull request as ready for review September 5, 2024 19:53

roblaszczak commented Sep 5, 2024

View reviewed changes

missing err check

bcd0160

Mariscal6 reviewed Sep 14, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[WIP] v1.0 preparations #17

[WIP] v1.0 preparations #17

roblaszczak commented Aug 20, 2024

roblaszczak Aug 23, 2024

Mariscal6 Aug 31, 2024

roblaszczak Sep 5, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024 •

edited

Loading

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

roblaszczak Aug 23, 2024

m110 Aug 26, 2024

m110 Aug 26, 2024

roblaszczak Sep 5, 2024

Mariscal6 commented Aug 31, 2024

roblaszczak Sep 5, 2024

Mariscal6 Sep 14, 2024

roblaszczak Sep 5, 2024

Mariscal6 Sep 14, 2024

Mariscal6 Sep 14, 2024

Mariscal6 Sep 14, 2024

Mariscal6 Sep 14, 2024

[WIP] v1.0 preparations #17

Are you sure you want to change the base?

[WIP] v1.0 preparations #17

Conversation

roblaszczak commented Aug 20, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roblaszczak Aug 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Mariscal6 commented Aug 31, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roblaszczak Aug 23, 2024 •

edited

Loading