circle ci

Author: qhenkart

.circleci/config.yml

@@ -0,0 +1,16 @@
+version: 2
+jobs:
+  build:
+    docker:
+      # specify the version
+      - image: circleci/golang:1.14
+      - image: pafortin/goaws
+    working_directory: /go/src/github.com/qhenkart/gosqs
+    steps:
+      - checkout
+      - run:
+        name: Install dependencies
+        command: |
+          go mod tidy
+          docker run --rm -v /go/src/github.com/qhenkart/gosqs/emulator/sqsconf.yaml:/sqsconf.yaml -p 4100:4100 pafortin/goaws --config /sqsconf.yaml
+      - run: go test -v ./...

consumer.go

@@ -100,6 +100,14 @@ func NewConsumer(c Config, queueName string) (Consumer, error) {
 	return cons, nil
 }
 
+// Logger accesses the logging field or applies a default logger
+func (c *consumer) Logger() Logger {
+	if c.logger == nil {
+		return &defaultLogger{}
+	}
+	return c.logger
+}
+
 // RegisterHandler registers an event listener and an associated handler. If the event matches, the handler will
 // be run along with any included middleware
 func (c *consumer) RegisterHandler(name string, h Handler, adapters ...Adapter) {
@@ -142,15 +150,15 @@ func (c *consumer) Consume() {
 	for {
 		output, err := c.sqs.ReceiveMessage(&sqs.ReceiveMessageInput{QueueUrl: &c.QueueURL, MaxNumberOfMessages: &maxMessages, MessageAttributeNames: []*string{&all}})
 		if err != nil {
-			c.logger.Println("%s , retrying in 10s", ErrGetMessage.Context(err).Error())
+			c.Logger().Println("%s , retrying in 10s", ErrGetMessage.Context(err).Error())
 			time.Sleep(10 * time.Second)
 			continue
 		}
 
 		for _, m := range output.Messages {
 			if _, ok := m.MessageAttributes["route"]; !ok {
 				//a message will be sent to the DLQ automatically after 4 tries if it is received but not deleted
-				c.logger.Println(ErrNoRoute.Error())
+				c.Logger().Println(ErrNoRoute.Error())
 				continue
 			}
 
@@ -163,7 +171,7 @@ func (c *consumer) Consume() {
 func (c *consumer) worker(id int, messages <-chan *message) {
 	for m := range messages {
 		if err := c.run(m); err != nil {
-			c.logger.Println(err.Error())
+			c.Logger().Println(err.Error())
 		}
 	}
 }
@@ -251,7 +259,7 @@ func (c *consumer) sendDirectMessage(ctx context.Context, input *sqs.SendMessage
 func (c *consumer) delete(m *message) error {
 	_, err := c.sqs.DeleteMessage(&sqs.DeleteMessageInput{QueueUrl: &c.QueueURL, ReceiptHandle: m.ReceiptHandle})
 	if err != nil {
-		c.logger.Println(ErrUnableToDelete.Context(err).Error())
+		c.Logger().Println(ErrUnableToDelete.Context(err).Error())
 		return ErrUnableToDelete.Context(err)
 	}
 	return nil
@@ -263,7 +271,7 @@ func (c *consumer) extend(ctx context.Context, m *message) {
 	for {
 		//only allow 1 extensions (Default 1m30s)
 		if count >= c.extensionLimit {
-			c.logger.Println(ErrMessageProcessing.Error(), m.Route())
+			c.Logger().Println(ErrMessageProcessing.Error(), m.Route())
 			return
 		}
 
@@ -279,7 +287,7 @@ func (c *consumer) extend(ctx context.Context, m *message) {
 			extension = extension + int64(c.VisibilityTimeout)
 			_, err := c.sqs.ChangeMessageVisibility(&sqs.ChangeMessageVisibilityInput{QueueUrl: &c.QueueURL, ReceiptHandle: m.ReceiptHandle, VisibilityTimeout: &extension})
 			if err != nil {
-				c.logger.Println(ErrUnableToExtend.Error(), err.Error())
+				c.Logger().Println(ErrUnableToExtend.Error(), err.Error())
 				return
 			}
 		}

consumer_test.go

@@ -44,8 +44,8 @@ func getConsumer(t *testing.T) *consumer {
 		Key:      "key",
 		Secret:   "secret",
 		Env:      "dev",
-		Hostname: "http://localhost:4150",
-		QueueURL: "http://goaws:4150/queue/dev-post-worker",
+		Hostname: "http://localhost:4100",
+		QueueURL: "http://local.goaws:4100/queue/dev-post-worker",
 	}
 	sess, err := newSession(conf)
 	if err != nil {
@@ -71,14 +71,14 @@ func TestNewConsumer(t *testing.T) {
 		Region:   "us-west2",
 		Key:      "key",
 		Secret:   "secret",
-		Hostname: "http://localhost:4150",
+		Hostname: "http://localhost:4100",
 		Env:      "dev",
 	}
 	c, err := NewConsumer(conf, "post-worker")
 	if err != nil {
 		t.Fatalf("error creating consumer, got %v", err)
 	}
-	expected := "http://goaws:4150/queue/dev-post-worker"
+	expected := "http://local.goaws:4100/queue/dev-post-worker"
 	if c.(*consumer).QueueURL != expected {
 		t.Fatalf("did not properly apply http result, expected %s, got %s", expected, c.(*consumer).QueueURL)
 	}

emulator/sqsconf.yaml

@@ -3,7 +3,7 @@ Local:                              # Environment name that can be passed on the
   Host: goaws                   # hostname of the goaws system  (for docker-compose this is the tag name of the container)
 # you can now use either 1 port for both sns and sqs or alternatively you can comment out Port and use SqsPort + SnsPort for compatibilyt with
 # yopa and (fage-sns + face-sqs).  If both ways are in the config file on the one "Port" will be used by GoAws
-  Port: 4150                        # port to listen on.
+  Port: 4100                        # port to listen on.
   Region: local
   LogMessages: true                 # Log messages (true/false)
   LogFile: ./goaws_messages.log  # Log filename (for message logging

go.mod

@@ -0,0 +1,5 @@
+module github.com/qhenkart/gosqs
+
+go 1.15
+
+require github.com/aws/aws-sdk-go v1.34.13

go.sum

@@ -0,0 +1,23 @@
+github.com/aws/aws-sdk-go v1.34.13 h1:wwNWSUh4FGJxXVOVVNj2lWI8wTe5hK8sGWlK7ziEcgg=
+github.com/aws/aws-sdk-go v1.34.13/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
+github.com/jmespath/go-jmespath v0.3.0 h1:OS12ieG61fsCg5+qLJ+SsW9NicxNkg3b25OyT2yCeUc=
+github.com/jmespath/go-jmespath v0.3.0/go.mod h1:9QtRXoHjLGCJ5IBSaohpXITPlowMeeYCZ7fLUTSywik=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

publisher_test.go

@@ -23,8 +23,8 @@ func TestNewPublisher(t *testing.T) {
 			Region:   "us-west-1",
 			Key:      "key",
 			Secret:   "secret",
-			Hostname: "http://localhost:4150",
-			TopicARN: "arn:aws:sns:local:000000000000:dev",
+			Hostname: "http://localhost:4100",
+			TopicARN: "arn:aws:sns:local:000000000000:todolist-dev",
 		}
 		_, err := NewPublisher(conf)
 		if err != nil {
@@ -38,17 +38,17 @@ func TestNewPublisher(t *testing.T) {
 			Key:          "key",
 			Secret:       "secret",
 			Env:          "dev",
-			Hostname:     "http://localhost:4150",
+			Hostname:     "http://localhost:4100",
 			AWSAccountID: "000000000000",
-			TopicPrefix:  "pre",
+			TopicPrefix:  "todolist",
 		}
 		pub, err := NewPublisher(conf)
 		if err != nil {
 			t.Fatalf("error creating publisher, got %v", err)
 		}
 		arn := pub.(*publisher).arn
-		if arn != "arn:aws:sns:local:000000000000:pre-dev" {
-			t.Errorf("did not properly create the arn name, expected %s, got %s", "arn:aws:sns:local:000000000000:pre-dev", arn)
+		if arn != "arn:aws:sns:local:000000000000:todolist-dev" {
+			t.Errorf("did not properly create the arn name, expected %s, got %s", "arn:aws:sns:local:000000000000:todolist-dev", arn)
 		}
 	})
 }
@@ -78,8 +78,8 @@ func getPublisher(t *testing.T) *publisher {
 		Key:      "key",
 		Env:      "dev",
 		Secret:   "secret",
-		Hostname: "http://localhost:4150",
-		TopicARN: "arn:aws:sns:local:000000000000:dev",
+		Hostname: "http://localhost:4100",
+		TopicARN: "arn:aws:sns:local:000000000000:todolist-dev",
 	}
 
 	sess, err := newSession(conf)