Adds out-of-the-box support for SQS message broker

deps.edn

@@ -10,7 +10,10 @@
            ring/ring-core                        {:mvn/version "1.11.0"}
            hiccup/hiccup                         {:mvn/version "2.0.0-RC2"}
            bidi/bidi                             {:mvn/version "2.1.6"}
-           it.burning/cron-expression-descriptor {:mvn/version "1.2.10"}}
+           it.burning/cron-expression-descriptor {:mvn/version "1.2.10"}
+           com.cognitect.aws/api       {:mvn/version "0.8.710-beta01"}
+           com.cognitect.aws/endpoints {:mvn/version "1.1.12.772"}
+           com.cognitect.aws/sqs       {:mvn/version "869.2.1687.0"}}
  :paths   ["src" "resources/public"]
 
  :aliases {:test       {:extra-paths ["test"]

src/goose/brokers/sqs/broker.clj

@@ -0,0 +1,82 @@
+(ns goose.brokers.sqs.broker
+  (:require
+   [goose.broker :as b]
+   [goose.brokers.sqs.worker :as sqs-worker]
+   [goose.brokers.sqs.sqs-requests :as sqs-request]
+   [cognitect.aws.client.api :as aws]
+   [goose.utils :as u]))
+
+(defprotocol Close
+  "Closes connection to SQS Message Broker."
+  (close [this]))
+
+(defrecord SQS [client opts]
+  b/Broker
+  ;; SQS supports nothing more than:
+  ;;   - Enqueue a job
+  ;;   - Start a worker process
+  ;;   - Enqueued jobs API
+  ;;   - Batch jobs (Will implement later)
+
+  (enqueue
+    [this job]
+    (let [{:keys [client]} this
+          {:keys [queue-url]} opts]
+
+      (sqs-request/enqueue client queue-url (u/encode-to-str job))))
+
+  (start-worker
+    [this worker-opts]
+    (let [{:keys [client]} this
+          {:keys [queue-url]} opts]
+      (sqs-worker/start client queue-url worker-opts)))
+
+  Close
+  (close [_]
+    (println "SQS client does not require explicit closure.")))
+
+(def default-opts
+  "Default Config is not allowed like redis or rmq
+   
+   ### Keys
+   `:region`: AWS Region
+   Example: \"ap-south-1\" , \"us-east-1\"
+   
+   `:queue-url`: The Queue URL provided by the AWS SQS
+   Example: \"https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue\""
+  {:region nil
+   :queue-url nil})
+
+(defn- create-sqs
+  "Helper function that creates an SQS broker implementation.
+  It abstracts the common logic for both producer and consumer."
+
+  [opts]
+  (let [client (aws/client {:api :sqs :region (:region opts)})]
+    (->SQS client opts)))
+
+(defn new-producer
+  "Creates an SQS broker implementation for producer (client).
+
+  ### Args
+  `opts` : Map of AWS client options.
+
+  ### Usage
+  ```clojure
+  (new-producer {:region \"us-east-1\" :queue-url \"https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue\"})
+  ```"
+  [opts]
+  (create-sqs opts))
+
+(defn new-consumer
+  "Creates an SQS broker implementation for worker (consumer).
+
+  ### Args
+  `opts` : Map of AWS client options.
+
+  ### Usage
+  ```clojure
+  (new-consumer {:region \"us-east-1\" :queue-url \"https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue\"})
+  ```"
+  [opts]
+  (create-sqs opts))

src/goose/brokers/sqs/sqs_requests.clj

@@ -0,0 +1,12 @@
+(ns goose.brokers.sqs.sqs-requests
+  (:require
+   [cognitect.aws.client.api :as aws]
+   [clojure.data.json :as json]))
+
+(defn enqueue
+  "Enqueues a job for immediate execution."
+  [client queue-url job]
+  (aws/invoke client {:op      :SendMessage
+                      :request {:QueueUrl     queue-url
+                                :MessageBody  (json/write-str job)
+                                :DelaySeconds 0}}))

src/goose/brokers/sqs/worker.clj

@@ -0,0 +1,37 @@
+(ns goose.brokers.sqs.worker
+  (:require
+   [cognitect.aws.client.api :as aws]
+   [clojure.core.async :as async]
+   [goose.utils :as u]
+   [goose.consumer :as c]))
+
+(defn start
+  "Polls messages from an SQS queue, executes each message as a job, and deletes it upon success.
+  
+  Args:
+  `client` : AWS SQS client.
+  `queue-url` : URL of the SQS queue.
+  `opts` : Additional options required for job execution."
+  [client queue-url opts]
+  (let [continue? (atom true)]
+    (async/go-loop []
+      (when @continue?
+        (let [response (aws/invoke client {:op :ReceiveMessage
+                                           :request {:QueueUrl queue-url
+                                                     :MaxNumberOfMessages 1
+                                                     :WaitTimeSeconds 10}})]
+          (when-let [messages (:Messages response)]
+            (doseq [message messages]
+
+              (let [{:keys [Body ReceiptHandle]} message
+                    job (u/decode-from-str Body)]
+                (try
+                  (c/execute-job opts job)
+                  (aws/invoke client {:op :DeleteMessage
+                                      :request {:QueueUrl queue-url
+                                                :ReceiptHandle ReceiptHandle}})
+                  (catch Exception e
+                    (println "Error executing job:" e))))))
+          (recur))))
+    (fn stop-polling []
+      (reset! continue? false))))