CMR-10212: SNS/SQS testing infrastructure (#2181)

* CMR-10212: Adding in SNS/SQS testing infrastructure

* CMR-10212: Adding in SNS/SQS testing infrastructure to dev-system

* CMR-10212: Adding in SNS/SQS adding in region to the clients.

* CMR-10212: Fixing PR requests and a bamboo library issue.

Author: eereiter

dev-system/src/cmr/dev_system/config.clj

@@ -70,3 +70,9 @@
   use an external instance."
   {:default :in-memory
    :parser parse-dev-system-component-type})
+
+(defconfig dev-system-sqs-server-type
+  "Specifies whether dev system should run an in-memory SQS service
+  (elasticmq) or use an external instance."
+  {:default :in-memory
+   :parser parse-dev-system-component-type})

dev-system/src/cmr/dev_system/system.clj

@@ -6,7 +6,7 @@
    [cmr.bootstrap.system :as bootstrap-system]
    [cmr.common.jobs :as jobs]
    [cmr.common.lifecycle :as lifecycle]
-   [cmr.common.log :refer [debug info warn error]]
+   [cmr.common.log :refer [info warn error]]
    [cmr.common.system :as common-sys]
    [cmr.common.util :as u]
    [cmr.dev-system.config :as dev-config]
@@ -21,6 +21,7 @@
    [cmr.message-queue.config :as rmq-conf]
    [cmr.message-queue.queue.memory-queue :as mem-queue]
    [cmr.message-queue.queue.sqs :as sqs]
+   [cmr.message-queue.queue-server.embedded-sqs-server :as sqs-server]
    [cmr.message-queue.test.queue-broker-wrapper :as wrapper]
    [cmr.metadata-db.data.memory-db :as memory]
    [cmr.metadata-db.system :as mdb-system]
@@ -46,7 +47,7 @@
                      peek)]
       (info "Using system token" token)
       token)
-    (catch Exception e
+    (catch Exception _e
       (warn "Unable to extract the ECHO system read token from configuration.")
       transmit-config/mock-echo-system-token)))
 
@@ -134,17 +135,25 @@
   [_]
   nil)
 
+(defn create-sqs-server
+  "Sets sqs server configuration values and returns an instance of a sqs-server component to run
+  in memory if applicable."
+  [type]
+  (case type
+    :in-memory (sqs-server/create-sqs-server)
+    nil))
+
 (defmulti create-db
   "Returns an instance of the database component to use."
   (fn [type]
     type))
 
 (defmethod create-db :in-memory
-  [type]
+  [_type]
   (memory/create-db))
 
 (defmethod create-db :external
-  [type]
+  [_type]
   nil)
 
 (defmulti create-echo
@@ -153,13 +162,15 @@
   (fn [type]
     type))
 
+#_{:clj-kondo/ignore [:unresolved-var]}
 (defmethod create-echo :in-memory
-  [type]
+  [_type]
   (transmit-config/set-urs-relative-root-url! "/urs")
   (mock-echo-system/create-system))
 
+#_{:clj-kondo/ignore [:unresolved-var]}
 (defmethod create-echo :external
-  [type]
+  [_type]
   (transmit-config/set-echo-rest-port! (dev-config/external-echo-port))
   (transmit-config/set-echo-system-token! (external-echo-system-token))
   (transmit-config/set-echo-rest-context! "/echo-rest"))
@@ -171,7 +182,7 @@
     type))
 
 (defmethod create-queue-broker :in-memory
-  [type]
+  [_type]
   (-> (indexer-config/queue-config)
       (rmq-conf/merge-configs (vp-config/queue-config))
       (rmq-conf/merge-configs (access-control-config/queue-config))
@@ -191,7 +202,7 @@
       (assoc :ttls ttls)))
 
 (defmethod create-queue-broker :aws
-  [type]
+  [_type]
   (-> (external-queue-config [])
       sqs/create-queue-broker))
 
@@ -230,18 +241,18 @@
 
 (defmulti create-ingest-app
   "Create an instance of the ingest application."
-  (fn [db-type queue-broker]
+  (fn [db-type _queue-broker]
     db-type))
 
 (defmethod create-ingest-app :in-memory
-  [db-type queue-broker]
+  [_db-type queue-broker]
   (assoc (ingest-system/create-system)
          :db (ingest-data/create-db)
          :queue-broker queue-broker
          :scheduler (jobs/create-non-running-scheduler)))
 
 (defmethod create-ingest-app :external
-  [db-type queue-broker]
+  [_db-type queue-broker]
   (assoc (ingest-system/create-system)
          :queue-broker queue-broker))
 
@@ -264,17 +275,19 @@
    :echo (dev-config/dev-system-echo-type)
    :db (dev-config/dev-system-db-type)
    :message-queue (dev-config/dev-system-queue-type)
-   :redis (dev-config/dev-system-redis-type)})
+   :redis (dev-config/dev-system-redis-type)
+   :sqs-server (dev-config/dev-system-sqs-server-type)})
 
 (defn create-system
   "Returns a new instance of the whole application."
   []
-  (let [{:keys [elastic echo db message-queue redis]} (component-type-map)
+  (let [{:keys [elastic echo db message-queue redis sqs-server]} (component-type-map)
         db-component (create-db db)
         echo-component (create-echo echo)
         queue-broker (create-queue-broker message-queue)
         elastic-server (create-elastic elastic)
         redis-server (create-redis redis)
+        sqs-server (create-sqs-server sqs-server)
         control-server (control/create-server)]
     {:instance-name (common-sys/instance-name "dev-system")
      :apps (u/remove-nil-keys
@@ -289,7 +302,8 @@
      :pre-components (u/remove-nil-keys
                        {:elastic-server elastic-server
                         :broker-wrapper queue-broker
-                        :redis-server redis-server})
+                        :redis-server redis-server
+                        :sqs-server sqs-server})
      :post-components {:control-server control-server}}))
 
 (defn- stop-components

message-queue-lib/project.clj

@@ -2,6 +2,10 @@
   "The java aws sdk version to use."
   "1.12.663")
 
+(def aws-java-sdk2-version
+  "The java aws sdk version to use."
+  "2.28.19")
+
 (defproject nasa-cmr/cmr-message-queue-lib "0.1.0-SNAPSHOT"
   :description "Library containing code to handle message queue interactions within the CMR."
   :url "https://github.com/nasa/Common-Metadata-Repository/tree/master/message-queue-lib"
@@ -10,6 +14,10 @@
                  [clj-time "0.15.1"]
                  [com.amazonaws/aws-java-sdk-sns ~aws-java-sdk-version]
                  [com.amazonaws/aws-java-sdk-sqs ~aws-java-sdk-version]
+                 [software.amazon.awssdk/regions ~aws-java-sdk2-version]
+                 [software.amazon.awssdk/sns ~aws-java-sdk2-version]
+                 [software.amazon.awssdk/sqs ~aws-java-sdk2-version]
+                 [com.fasterxml.jackson.core/jackson-annotations "2.15.4"]
                  [commons-codec/commons-codec "1.11"]
                  [commons-io "2.6"]
                  [commons-logging "1.2"]
@@ -21,6 +29,7 @@
                  [org.apache.httpcomponents/httpcore "4.4.10"]
                  [org.clojure/clojure "1.11.2"]
                  [org.clojure/tools.reader "1.3.2"]
+                 [org.testcontainers/testcontainers "1.19.7"]
                  [potemkin "0.4.5"]]
   :plugins [[lein-shell "0.5.0"]]
   :jvm-opts ^:replace ["-server"

message-queue-lib/src/cmr/message_queue/config.clj

@@ -63,3 +63,52 @@
                        q-to-e
                        (:queues-to-exchanges config2))))
         (update-in [:queues-to-policies] merge (:queues-to-policies config2)))))
+
+(declare cmr-internal-subscriptions-topic-name)
+(defconfig cmr-internal-subscriptions-topic-name
+  "The name of the internal subscriptions topic."
+  {:default (str "cmr-internal-subscriptions-" (app-environment))})
+
+(declare cmr-subscriptions-topic-name)
+(defconfig cmr-subscriptions-topic-name
+  "The name of the subscriptions topic to which external clients can subscribe."
+  {:default (str "cmr-subscriptions-" (app-environment))})
+
+(declare cmr-internal-subscriptions-queue-name)
+(defconfig cmr-internal-subscriptions-queue-name
+  "The name of the internal subscriptions queue."
+  {:default (str "cmr-internal-subscriptions-queue-" (app-environment))})
+
+(declare cmr-internal-subscriptions-dead-letter-queue-name)
+(defconfig cmr-internal-subscriptions-dead-letter-queue-name
+  "The name of the internal subscriptions queue."
+  {:default (str "cmr-internal-subscriptions-dead-letter-queue-" (app-environment))})
+
+(declare cmr-subscriptions-dead-letter-queue-name)
+(defconfig cmr-subscriptions-dead-letter-queue-name
+  "The name of the subscriptions dead letter queue for external subscriptions."
+  {:default (str "cmr-subscriptions-dead-letter-queue-" (app-environment))})
+
+(declare sqs-server-protocol)
+(defconfig sqs-server-protocol
+  "Default SQS server host."
+  {:default "http"})
+
+(declare sqs-server-host)
+(defconfig sqs-server-host
+  "Default SQS server host."
+  {:default "localhost"})
+
+(declare sqs-server-port)
+(defconfig sqs-server-port
+  "Default SQS server host."
+  {:default "9324"})
+
+(declare sqs-server-url)
+(defconfig sqs-server-url
+  "The URL of the local or test SQS implementation."
+  {:default (str (sqs-server-protocol)
+                 "://"
+                 (sqs-server-host)
+                 ":"
+                 (sqs-server-port))})

message-queue-lib/src/cmr/message_queue/pub_sub.clj

@@ -0,0 +1,15 @@
+(ns cmr.message-queue.pub-sub 
+  (:require
+   [cmr.message-queue.config :as config]
+   #_{:clj-kondo/ignore [:unused-namespace]}
+   [cmr.message-queue.topic.aws-topic :as aws-topic]
+   [cmr.message-queue.topic.local-topic :as local-topic]))
+
+(defn create-topic
+  "Create a topic using the given topic configuration. The type is determined
+    by the environment variable CMR_QUEUE_TYPE."
+  []
+  (let [create-fn (case (config/queue-type)
+                    "memory" local-topic/setup-topic nil
+                    "aws" aws-topic/setup-topic (config/cmr-internal-subscriptions-topic-name))]
+    (create-fn)))