Merge pull request #34 from meero-com/feat/configurable-backend

Configurable Backend

Author: tbobm

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_ddb.sh

@@ -0,0 +1,45 @@
+#!/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)
+
+# Background process: posts item to DynamoDB 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 dynamodb put-item \
+    --endpoint-url http://localhost:4566 \
+    --table-name response-ddb-table \
+    --item '{"uuid": {"S": "'$uuid'"}, "payload": { "M": { "name": { "S": "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."

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

@@ -1,33 +1,40 @@
 package api
 
 import (
-	"context"
 	"log"
 
-	"github.com/google/uuid"
 	"github.com/meero-com/hmb-proxy/pkg/aws"
 	"github.com/meero-com/hmb-proxy/pkg/config"
 	"github.com/meero-com/hmb-proxy/pkg/pollers"
+	"github.com/meero-com/hmb-proxy/pkg/producers"
 )
 
 func process(ch chan string, p requestPayload) {
-	ddb := aws.NewDdbCoordinator()
-	uuid := uuid.NewString()
-	requestTable := config.GetConfig("ddb.request_table").(string)
+	var producer producers.Producer
+	var poller pollers.Poller
 
-	ddbPayload := aws.DdbPayload{
-		Name: p.Payload.Name,
-	}
-	ddbi := aws.DdbItem{
-		Uuid:    p.Uuid,
-		Payload: ddbPayload,
-	}
-
-	_, err := ddb.Put(context.Background(), requestTable, ddbi)
+	backendType := config.GetConfig("backend_type").(string)
 
-	if err != nil {
-		log.Fatalf("Failed to put item %s into table %s", uuid, requestTable)
+	if backendType == "ddb" {
+		ddb := aws.NewDdbCoordinator()
+		producer = &producers.DdbProducer{
+			Ddb: ddb,
+		}
+		poller = &pollers.DdbPoller{
+			Ddb: ddb,
+		}
+	} else if backendType == "sqs" {
+		sqs := aws.NewSqsCoordinator()
+		producer = &producers.SqsProducer{
+			Sqs: sqs,
+		}
+		poller = &pollers.SqsPoller{
+			Sqs: sqs,
+		}
+	} else {
+		log.Fatalf("Unsupported backend type: %s", backendType)
 	}
 
-	pollers.PollDdb(ch, p.Uuid, ddb)
+	producer.Produce(p.Uuid, p.Payload.Name)
+	poller.Poll(ch, p.Uuid)
 }

pkg/config/config.yaml

@@ -3,11 +3,16 @@ env: dev
 server:
   port: 8080
 
+backend_type: sqs
+
 aws:
   access_key_id: default
   secret_access_key: default
   region: eu-west-1
   endpoint_url: http://localstack:4566/
 sqs:
   source_queue: input-queue
-  destination_queue: output-queue
+  destination_queue: result-queue
+ddb:
+  request-table: request-ddb-table
+  response-table: response-ddb-table

pkg/pollers/ddb_poller.go

@@ -11,11 +11,12 @@ import (
 	"github.com/meero-com/hmb-proxy/pkg/config"
 )
 
-const (
-	pollInterval = 6 * time.Second
-)
+type DdbPoller struct {
+	Ddb aws.DdbCoordinator
+}
 
-func PollDdb(ch chan string, uuid string, ddb aws.DdbCoordinator) {
+func (p *DdbPoller) Poll(ch chan string, uuid string) {
+	ddb := p.Ddb
 	responseTable := config.GetConfig("ddb.response_table").(string)
 	ctx := context.Background()
 

pkg/pollers/model.go

@@ -1 +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)
+	}
+}