Skip to content

Commit 7ba4da3

Browse files
jmaeng72jmaeng72
committed
update subscripion message attributes
1 parent 9d49d71 commit 7ba4da3

File tree

5 files changed

+148
-137
lines changed

5 files changed

+148
-137
lines changed

‎ingest-app/src/cmr/ingest/api/subscriptions.clj‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -333,7 +333,7 @@
333333
parsed)]
334334
(when-not (= CMR_PROVIDER provider-id)
335335
(api-core/verify-provider-exists context provider-id))
336-
;(validate-user-id context subscriber-id) ;; TODO temp comment out to bypass user id check
336+
(validate-user-id context subscriber-id) ;; TODO temp comment out to bypass user id check
337337
(validate-query context parsed)
338338
(validate-subscription-endpoint parsed)
339339
(let [parsed-metadata (assoc parsed :SubscriberId subscriber-id)]
@@ -348,17 +348,14 @@
348348
"Processes a request to create a subscription. A native id will be generated."
349349
[request]
350350
(let [{:keys [body content-type headers request-context]} request]
351-
(info "**** INSIDE create-subscription")
352351
(common-ingest-checks request-context)
353352
(let [tmp-subscription (body->subscription (str (UUID/randomUUID)) body content-type headers)
354353
{:keys [concept parsed]} (validate-and-prepare-subscription-concept
355354
request-context tmp-subscription)
356-
_ (info "**** concept = " concept)
357355
provider-id (:provider-id concept)
358356
subscriber-id (:SubscriberId parsed)
359357
native-id (get-unique-native-id request-context parsed)
360-
final-sub (assoc concept :native-id native-id)
361-
_ (info "**** final-sub = " final-sub)]
358+
final-sub (assoc concept :native-id native-id)]
362359
(check-ingest-permission request-context provider-id subscriber-id)
363360
(perform-subscription-ingest request-context headers final-sub parsed))))
364361

‎metadata-db-app/src/cmr/metadata_db/services/concept_service.clj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -914,7 +914,7 @@
914914
(subscriptions/set-subscription-arn-if-applicable context concept-type) ;; TODO this is where the subscription stuff happens
915915
(set-or-generate-revision-id db provider)
916916
(set-deleted-flag false)
917-
(try-to-save db provider context)) ;; saving the subscription into the db here
917+
(try-to-save db provider context))
918918
revision-id (:revision-id concept)]
919919
;; publish tombstone delete event if the previous concept revision is a granule tombstone
920920
(when (and (= :granule concept-type)

‎metadata-db-app/src/cmr/metadata_db/services/subscription_cache.clj‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,7 @@
2727
(defn set-value
2828
"Set the collection concept id and its subscription map described at the top."
2929
[context field value]
30-
(let [_ (println (format "**** INSIDE set-value with field = %s and value = %s" field value))
31-
cache-client (hash-cache/context->cache context subscription-cache-key)
30+
(let [cache-client (hash-cache/context->cache context subscription-cache-key)
3231
[tm result] (util/time-execution
3332
(hash-cache/set-value cache-client subscription-cache-key field value))]
3433
(rl-util/log-redis-write-complete "ingest-subscription-cache set-value" subscription-cache-key tm)

‎metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj‎

Lines changed: 89 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -84,45 +84,6 @@
8484
result
8585
(recur (rest subs) (add-to-existing-mode result (get-in sub [:metadata :Mode])))))))
8686

87-
"
88-
({
89-
:revision-id 1,
90-
:deleted false,
91-
:format application/vnd.nasa.cmr.umm+json;version=1.1.1,
92-
:provider-id JM_PROV1,
93-
:subscription-type granule,
94-
:user-id ECHO_SYS,
95-
:transaction-id 5,
96-
:native-id jyna_ingest_subscription_23,
97-
:normalized-query bca0dee3632df33a114bd85059accb71,
98-
:concept-id SUB1200000003-JM_PROV1,
99-
:created-at 2025-01-31T17:41:16.480Z,
100-
:metadata {
101-
:SubscriberId user2,
102-
:CollectionConceptId C1200000001-JM_PROV1,
103-
:EndPoint http://localhost:9324/000000000000/cmr-internal-subscriptions-queue-local,
104-
:Mode [Update],
105-
:EmailAddress jyna.maeng@nasa.gov,
106-
:Query collection-concept-id=C1200000001-JM_PROV1,
107-
:Name Ingest-Subscription-Test,
108-
:Method ingest,
109-
:Type granule,
110-
:MetadataSpecification {:URL https://cdn.earthdata.nasa.gov/umm/subscription/v1.1.1, :Name UMM-Sub, :Version 1.1.1}
111-
},
112-
:revision-date 2025-01-31T17:41:16.480Z,
113-
:extra-fields {
114-
:aws-arn SUB1200000003-JM_PROV1,
115-
:subscription-name Ingest-Subscription-Test,
116-
:method ingest,
117-
:collection-concept-id C1200000001-JM_PROV1,
118-
:subscriber-id user2,
119-
:subscription-type granule,
120-
:mode [Update],
121-
:normalized-query bca0dee3632df33a114bd85059accb71,
122-
:endpoint http://localhost:9324/000000000000/cmr-internal-subscriptions-queue-local
123-
},
124-
:concept-type :subscription
125-
}"
12687

12788
(use 'clojure.set)
12889
(defn create-mode-to-endpoints-map
@@ -334,7 +295,7 @@
334295
(:revision-id concept))
335296
"\""))
336297

337-
(defn create-notification
298+
(defn create-notification-message-body
338299
"Create the notification when a subscription exists. Returns either a notification message or nil."
339300
[concept]
340301
(let [concept-edn (convert-concept-to-edn concept)
@@ -356,53 +317,107 @@
356317
[mode]
357318
(str mode " Notification"))
358319

359-
(defn create-attributes-and-subject-map
360-
"Determine based on the passed in concept if the granule is new, is an update
361-
or delete. Use the passed in mode to determine if any subscription is interested
362-
in a notification. If they are then return the message attributes and subject, otherwise
363-
return nil."
364-
[concept mode coll-concept-id]
320+
;(defn create-attributes-and-subject-map
321+
; "Determine based on the passed in concept if the granule is new, is an update
322+
; or delete. Use the passed in mode to determine if any subscription is interested
323+
; in a notification. If they are then return the message attributes and subject, otherwise
324+
; return nil."
325+
; [concept mode coll-concept-id]
326+
; (cond
327+
; ;; Mode = Delete.
328+
; (and (:deleted concept)
329+
; (some #(= "Delete" %) mode))
330+
; {:attributes (create-message-attributes coll-concept-id "Delete")
331+
; :subject (create-message-subject "Delete")}
332+
;
333+
; ;; Mode = New
334+
; (and (not (:deleted concept))
335+
; (= 1 (:revision-id concept))
336+
; (some #(= "New" %) mode))
337+
; {:attributes (create-message-attributes coll-concept-id "New")
338+
; :subject (create-message-subject "New")}
339+
;
340+
; ;; Mode = Update
341+
; (and (not (:deleted concept))
342+
; (pos? (compare (:revision-id concept) 1))
343+
; (some #(= "Update" %) mode))
344+
; {:attributes (create-message-attributes coll-concept-id "Update")
345+
; :subject (create-message-subject "Update")}))
346+
347+
;(defn publish-subscription-notification-if-applicable
348+
; "Publish a notification to the topic if the passed-in concept is a granule
349+
; and a subscription is interested in being informed of the granule's actions."
350+
; [context concept]
351+
; (when (granule-concept? (:concept-type concept))
352+
; (let [start (System/currentTimeMillis)
353+
; coll-concept-id (:parent-collection-id (:extra-fields concept))
354+
; sub-cache-map (subscription-cache/get-value context coll-concept-id)]
355+
; ;; if this granule's collection is found in subscription cache that means it has a subscription attached to it
356+
; (when sub-cache-map
357+
; ;; Check the mode to see if the granule notification needs to be pushed. Mode examples are 'new', 'update', 'delete'.
358+
; (let [topic (get-in context [:system :sns :internal])
359+
; message (create-notification-message-body concept)
360+
; ;; TODO Jyna will need to update this attributes and subject map for URL endpoints
361+
; {:keys [attributes subject]} (create-attributes-and-subject-map concept sub-cache-map coll-concept-id)]
362+
; (when (and attributes subject)
363+
; (let [result (topic-protocol/publish topic message attributes subject)
364+
; duration (- (System/currentTimeMillis) start)]
365+
; (debug (format "Work potential subscription publish took %d ms." duration))
366+
; result)))))))
367+
368+
369+
(defn- get-gran-concept-mode
370+
[concept]
365371
(cond
366-
;; Mode = Delete.
367-
(and (:deleted concept)
368-
(some #(= "Delete" %) mode))
369-
{:attributes (create-message-attributes coll-concept-id "Delete")
370-
:subject (create-message-subject "Delete")}
371-
372+
(:deleted concept) "Delete"
372373
;; Mode = New
373-
(and (not (:deleted concept))
374-
(= 1 (:revision-id concept))
375-
(some #(= "New" %) mode))
376-
{:attributes (create-message-attributes coll-concept-id "New")
377-
:subject (create-message-subject "New")}
378-
374+
(and (not (:deleted concept)) (= 1 (:revision-id concept))) "New"
379375
;; Mode = Update
380-
(and (not (:deleted concept))
381-
(pos? (compare (:revision-id concept) 1))
382-
(some #(= "Update" %) mode))
383-
{:attributes (create-message-attributes coll-concept-id "Update")
384-
:subject (create-message-subject "Update")}))
376+
(and (not (:deleted concept)) (pos? (compare (:revision-id concept) 1))) "Update"
377+
))
378+
379+
(defn- create-message-attributes-map
380+
[endpoint mode coll-concept-id]
381+
(cond
382+
(or (is-valid-sqs-arn endpoint) (is-local-test-queue endpoint)) (cond
383+
(= "Delete" mode) {:attributes (create-message-attributes coll-concept-id "Delete")}
384+
(= "New" mode) {:attributes (create-message-attributes coll-concept-id "New")}
385+
(= "Update" mode) {:attributes (create-message-attributes coll-concept-id "Update")})
386+
(is-valid-subscription-endpoint-url endpoint) {:attributes {"endpoint" endpoint
387+
"endpoint-type" "url"
388+
"mode" mode}}))
385389

386390
(defn publish-subscription-notification-if-applicable
387391
"Publish a notification to the topic if the passed-in concept is a granule
388392
and a subscription is interested in being informed of the granule's actions."
389393
[context concept]
394+
(println "***** INSIDE publish-subscription-notification-if-applicable")
390395
(when (granule-concept? (:concept-type concept))
391396
(let [start (System/currentTimeMillis)
392397
coll-concept-id (:parent-collection-id (:extra-fields concept))
398+
_ (println "coll-concept-id = " coll-concept-id)
393399
sub-cache-map (subscription-cache/get-value context coll-concept-id)]
394-
;; if this granule's collection is found in subscription cache
400+
;; if this granule's collection is found in subscription cache that means it has a subscription attached to it
395401
(when sub-cache-map
396-
;; Check the mode to see if the granule notification needs to be pushed. Mode examples are 'new', 'update', 'delete'.
397-
(let [topic (get-in context [:system :sns :internal])
398-
message (create-notification concept)
399-
;; TODO Jyna will need to update this attributes and subject map for URL endpoints
400-
{:keys [attributes subject]} (create-attributes-and-subject-map concept sub-cache-map coll-concept-id)]
401-
(when (and attributes subject)
402-
(let [result (topic-protocol/publish topic message attributes subject)
403-
duration (- (System/currentTimeMillis) start)]
404-
(debug (format "Work potential subscription publish took %d ms." duration))
405-
result)))))))
402+
(let [gran-concept-mode (get-gran-concept-mode concept)
403+
_ (println "gran-concept-mode = " gran-concept-mode)
404+
endpoint-list (get sub-cache-map gran-concept-mode)
405+
_ (println "endpoint-list = " endpoint-list)]
406+
;; for every endpoint in the list create a attributes/subject map and send it along its way
407+
(doseq [endpoint endpoint-list]
408+
(let [topic (get-in context [:system :sns :internal])
409+
coll-concept-id (:parent-collection-id (:extra-fields concept))
410+
message (create-notification-message-body concept)
411+
message-attributes-map (create-message-attributes-map endpoint gran-concept-mode coll-concept-id)
412+
_ (println "message-attributes-map = " message-attributes-map)
413+
subject-map {:subject (create-message-subject gran-concept-mode)}
414+
_ (println "subject map = " subject-map)]
415+
(when (and message-attributes-map subject-map)
416+
(let [result (topic-protocol/publish topic message message-attributes-map subject-map)
417+
duration (- (System/currentTimeMillis) start)]
418+
(debug (format "Subscription publish for endpoint %s took %d ms." endpoint duration))
419+
result)))
420+
))))))
406421

407422
(comment
408423
(let [system (get-in user/system [:apps :metadata-db])]

‎metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj‎

Lines changed: 55 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -55,12 +55,12 @@
5555
\"Method\":\"ingest\"}",
5656
:extra-fields {:collection-concept-id "C12345-PROV1"}}))}
5757
(is (= 1 (subscriptions/change-subscription-in-cache test-context {:concept-type :subscription
58-
:deleted false
59-
:metadata {:CollectionConceptId "C12345-PROV1"
58+
:deleted false
59+
:metadata {:CollectionConceptId "C12345-PROV1"
6060
:EndPoint "some-endpoint"
6161
:Mode ["New"]
6262
:Method "ingest"}
63-
:extra-fields {:collection-concept-id "C12345-PROV1"}})))))
63+
:extra-fields {:collection-concept-id "C12345-PROV1"}})))))
6464
(testing "Delete a subscription from the cache"
6565
(with-bindings {#'subscriptions/get-subscriptions-from-db (fn [_context _coll-concept-id] '({:concept-type :subscription
6666
:deleted true
@@ -70,12 +70,12 @@
7070
\"Method\":\"ingest\"}",
7171
:extra-fields {:collection-concept-id "C12345-PROV1"}}))}
7272
(is (= 1 (subscriptions/change-subscription-in-cache test-context {:concept-type :subscription
73-
:deleted true
74-
:metadata {:CollectionConceptId "C12345-PROV1"
73+
:deleted true
74+
:metadata {:CollectionConceptId "C12345-PROV1"
7575
:EndPoint "some-endpoint"
7676
:Mode ["New"]
7777
:Method "ingest"}
78-
:extra-fields {:collection-concept-id "C12345-PROV1"}})))))
78+
:extra-fields {:collection-concept-id "C12345-PROV1"}})))))
7979
(testing "Add-to-existing-mode"
8080
(are3
8181
[expected existing-modes new-modes]
@@ -235,10 +235,10 @@
235235
:Method "ingest"}
236236
:concept-type :subscription}
237237
(subscriptions/add-or-delete-ingest-subscription-in-cache test-context {:metadata "{\"CollectionConceptId\":\"C12345-PROV1\",
238-
\"EndPoint\":\"ARN\",
238+
\"EndPoint\":\"ARN\",
239239
\"Mode\":[\"New\", \"Delete\"],
240240
\"Method\":\"ingest\"}"
241-
:concept-type :subscription}))))))))
241+
:concept-type :subscription}))))))))
242242

243243
(def db-result-1
244244
'({:revision-id 1
@@ -406,7 +406,7 @@
406406
"\"granule-ur\": \"GranuleUR\", "
407407
"\"producer-granule-id\": \"Algorithm-1\", "
408408
"\"location\": \"http://localhost:3003/concepts/G12345-PROV1/1\"}")
409-
(subscriptions/create-notification concept))))))
409+
(subscriptions/create-notification-message-body concept))))))
410410

411411
(deftest create-message-attributes-test
412412
(testing "Creating the message attributes."
@@ -422,52 +422,52 @@
422422
(is (= "Delete Notification"
423423
(subscriptions/create-message-subject mode))))))
424424

425-
(deftest get-attributes-and-subject-test
426-
(testing "Getting notificaiton attributes and subject."
427-
(are3
428-
;;concept mode coll-concept-id
429-
[expected concept mode coll-concept-id]
430-
(is (= expected
431-
(subscriptions/create-attributes-and-subject-map concept mode coll-concept-id)))
432-
433-
"Deleted concept"
434-
{:attributes {"collection-concept-id" "C12345-PROV1"
435-
"mode" "Delete"}
436-
:subject "Delete Notification"}
437-
{:deleted true}
438-
["New" "Delete"]
439-
"C12345-PROV1"
440-
441-
"Deleted concept, but not looking for the mode."
442-
nil
443-
{:deleted true}
444-
["New" "Update"]
445-
"C12345-PROV1"
446-
447-
"Deleted concept, but not looking for the mode, making sure no other condition is met."
448-
nil
449-
{:deleted true
450-
:revision-id 2}
451-
["New" "Update"]
452-
"C12345-PROV1"
453-
454-
"New concept."
455-
{:attributes {"collection-concept-id" "C12345-PROV1"
456-
"mode" "New"}
457-
:subject "New Notification"}
458-
{:deleted false
459-
:revision-id 1}
460-
["New" "Update"]
461-
"C12345-PROV1"
462-
463-
"Update concept."
464-
{:attributes {"collection-concept-id" "C12345-PROV1"
465-
"mode" "Update"}
466-
:subject "Update Notification"}
467-
{:deleted false
468-
:revision-id 3}
469-
["New" "Update"]
470-
"C12345-PROV1")))
425+
;(deftest get-attributes-and-subject-test
426+
; (testing "Getting notificaiton attributes and subject."
427+
; (are3
428+
; ;;concept mode coll-concept-id
429+
; [expected concept mode coll-concept-id]
430+
; (is (= expected
431+
; (subscriptions/create-attributes-and-subject-map concept mode coll-concept-id)))
432+
;
433+
; "Deleted concept"
434+
; {:attributes {"collection-concept-id" "C12345-PROV1"
435+
; "mode" "Delete"}
436+
; :subject "Delete Notification"}
437+
; {:deleted true}
438+
; ["New" "Delete"]
439+
; "C12345-PROV1"
440+
;
441+
; "Deleted concept, but not looking for the mode."
442+
; nil
443+
; {:deleted true}
444+
; ["New" "Update"]
445+
; "C12345-PROV1"
446+
;
447+
; "Deleted concept, but not looking for the mode, making sure no other condition is met."
448+
; nil
449+
; {:deleted true
450+
; :revision-id 2}
451+
; ["New" "Update"]
452+
; "C12345-PROV1"
453+
;
454+
; "New concept."
455+
; {:attributes {"collection-concept-id" "C12345-PROV1"
456+
; "mode" "New"}
457+
; :subject "New Notification"}
458+
; {:deleted false
459+
; :revision-id 1}
460+
; ["New" "Update"]
461+
; "C12345-PROV1"
462+
;
463+
; "Update concept."
464+
; {:attributes {"collection-concept-id" "C12345-PROV1"
465+
; "mode" "Update"}
466+
; :subject "Update Notification"}
467+
; {:deleted false
468+
; :revision-id 3}
469+
; ["New" "Update"]
470+
; "C12345-PROV1")))
471471

472472
(defn set-db-result
473473
"Sets the mock db result with a real queue endpoint to

0 commit comments

Comments
 (0)