feat: sqs producer and poller

Author: Oliver Allweyer

docker-compose.yaml

@@ -20,7 +20,7 @@ services:
       AWS_REGION: eu-west-1
       AWS_ENDPOINT_URL: http://localstack:4566/
       SQS_SOURCE_QUEUE: input-queue
-      SQS_DESTINATION_QUEUE: output-queue
+      SQS_DESTINATION_QUEUE: result-queue
       DDB_REQUEST_TABLE: request-ddb-table
       DDB_RESPONSE_TABLE: response-ddb-table
     build:

dummy-service/envrc.dist

@@ -1,5 +1,5 @@
 export SQS_SOURCE_QUEUE=input-queue
-export SQS_DESTINATION_QUEUE=output-queue
+export SQS_DESTINATION_QUEUE=result-queue
 
 export AWS_ACCESS_KEY_ID=test
 export AWS_SECRET_ACCESS_KEY=test

e2e/test_sqs.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+
+set +e  # Disable exit on error, we want to clean up after an error
+
+export AWS_ACCESS_KEY_ID="test"
+export AWS_SECRET_ACCESS_KEY="test"
+export AWS_DEFAULT_REGION="eu-west-1"
+export AWS_ENDPOINT_URL="http://localhost:4566"
+
+uuid=$(uuidgen)
+
+# get the queue URL
+queue_url=$(aws sqs get-queue-url --queue-name result-queue | jq -r '.QueueUrl')
+echo "Queue URL: $queue_url"
+
+# Background process: posts item to SQS queue to trigger the response
+(
+  # Generate random number between 1-5
+  sleep_time=$(( (RANDOM % 5) + 1 ))
+  echo "Will execute AWS command after ${sleep_time} seconds..."
+  sleep ${sleep_time}
+  aws sqs send-message \
+    --endpoint-url http://localhost:4566 \
+    --queue-url http://localhost:4566/000000000000/result-queue \
+    --message-body "{\"uuid\": \"$uuid\", \"payload\": {\"name\": \"response payload from service\"}}"
+  echo "AWS command executed."
+) &
+background_pid=$!
+
+echo "Request UUID: $uuid"
+
+curl -XPOST --retry 5 --fail-with-body \
+    localhost:8080/api/process \
+    -H "Content-Type: application/json" \
+    -d "{\"uuid\": \"$uuid\", \"payload\": {\"name\": \"default\", \"timeout\": 20}}"
+curl_exit_status=$?
+
+if [ $curl_exit_status -ne 0 ]; then
+  echo "\nCurl failed with status $curl_exit_status. Killing background process..."
+  kill $background_pid
+  wait $background_pid 2>/dev/null || true
+  echo "Background process terminated."
+  echo "Test failed."
+  exit $curl_exit_status
+fi
+
+echo "\nWaiting for AWS command to complete..."
+wait $background_pid
+echo "Test successful."

pkg/api/process.go

@@ -23,9 +23,16 @@ func process(ch chan string, p requestPayload) {
 		poller = &pollers.DdbPoller{
 			Ddb: ddb,
 		}
+	} else if backendType == "sqs" {
+		sqs := aws.NewSqsCoordinator()
+		producer = &producers.SqsProducer{
+			Sqs: sqs,
+		}
+		poller = &pollers.SqsPoller{
+			Sqs: sqs,
+		}
 	} else {
-		// TODO: sqs producer and poller
-		log.Fatalf("SQS backend not yet implemented")
+		log.Fatalf("Unsupported backend type: %s", backendType)
 	}
 
 	producer.Produce(p.Uuid, p.Payload.Name)

pkg/config/config.yaml

@@ -3,7 +3,7 @@ env: dev
 server:
   port: 8080
 
-backend_type: ddb
+backend_type: sqs
 
 aws:
   access_key_id: default

pkg/pollers/ddb_poller.go

@@ -11,10 +11,6 @@ import (
 	"github.com/meero-com/hmb-proxy/pkg/config"
 )
 
-const (
-	pollInterval = 6 * time.Second
-)
-
 type DdbPoller struct {
 	Ddb aws.DdbCoordinator
 }

pkg/pollers/model.go

@@ -1,5 +1,11 @@
 package pollers
 
+import "time"
+
 type Poller interface {
 	Poll(ch chan string, uuid string)
 }
+
+const (
+	pollInterval = 6 * time.Second
+)

pkg/pollers/poller.go

@@ -0,0 +1,47 @@
+package pollers
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"time"
+
+	"github.com/meero-com/hmb-proxy/pkg/aws"
+	"github.com/meero-com/hmb-proxy/pkg/config"
+)
+
+type SqsPoller struct {
+	Sqs aws.SqsCoordinator
+}
+
+func (p *SqsPoller) Poll(ch chan string, uuid string) {
+	c := p.Sqs
+	destinationQueue := config.GetConfig("sqs.destination_queue").(string)
+
+	log.Printf("Start polling\n")
+
+	for {
+		messages, _ := c.GetMessages(context.Background(), destinationQueue, 10, 10)
+		log.Printf("got %d messages \n", len(messages))
+		for idx, message := range messages {
+			log.Printf("got messages #%d id=%s %s\n", idx, *message.MessageId, *message.Body)
+
+			// TODO: check if message attributes contain uuid, if not reschedule
+
+			err := c.AckMessage(context.Background(), destinationQueue, message.ReceiptHandle)
+			if err != nil {
+				log.Fatalf("Failed to acknowledge message %s", *message.MessageId)
+			}
+
+			si, err := json.Marshal(*message.Body)
+			if err != nil {
+				log.Fatalf("Failed to Marshal item: %s", *message.Body)
+			}
+			ch <- string(si)
+			close(ch)
+			return
+		}
+
+		time.Sleep(pollInterval)
+	}
+}

pkg/pollers/sqs_poller.go

@@ -0,0 +1,27 @@
+package producers
+
+import (
+	"context"
+	"log"
+
+	"github.com/meero-com/hmb-proxy/pkg/aws"
+	"github.com/meero-com/hmb-proxy/pkg/config"
+)
+
+type SqsProducer struct {
+	Sqs aws.SqsCoordinator
+}
+
+func (p *SqsProducer) Produce(uuid string, payload string) {
+	sqs := p.Sqs
+	queueUrl := config.GetConfig("sqs.source_queue").(string)
+
+	// TODO: forge sqs payload with uuid in message attributes
+	sqsPayload := payload
+
+	err := sqs.PutMessage(context.Background(), queueUrl, &sqsPayload)
+
+	if err != nil {
+		log.Fatalf("Failed to put message %s to sqs queue %s", uuid, queueUrl)
+	}
+}