Merge pull request #114 from CDCgov/dev-feature-add-listeners-sb-topics

Add listeners sb topics, Reports DeadLetter and Remove ContentAsString

processing-status-api-function-app/gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
+#Fri Jun 07 10:39:15 EDT 2024
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.0.2-bin.zip
-networkTimeout=10000
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

pstatus-report-sink-ktor/README.md

@@ -1,5 +1,5 @@
 # Overview
-This project is the processing status report sink.  It listens for messages on an Azure Service bus, validates the messages, and if validated persists them to CosmosDB.
+This project is the processing status report sink.  It listens for messages on an Azure Service bus queues and topics, validates the messages, and if validated persists them to CosmosDB. If the validation fails due to missing fields or malformed data, then the message is persisted in cosmosdb under a new dead-letter container and the message is also sent to the dead-letter queue under the configured topic subscription(if the message was processed using the topic listener) 
 
 This is a microservice built using Ktor that can be built as a docker container image.
 

pstatus-report-sink-ktor/build.gradle

@@ -45,6 +45,52 @@ dependencies {
     implementation("io.insert-koin:koin-ktor:3.5.6")
     testImplementation("io.ktor:ktor-server-tests-jvm")
     testImplementation("org.jetbrains.kotlin:kotlin-test-junit:$kotlin_version")
+
+    implementation "com.microsoft.azure.functions:azure-functions-java-library:3.0.0"
+    implementation 'com.microsoft.azure:applicationinsights-core:3.4.19'
+    implementation 'com.azure:azure-cosmos:4.55.0'
+    implementation "com.fasterxml.jackson.module:jackson-module-kotlin:2.14.2"
+    implementation 'io.opentelemetry:opentelemetry-api:1.29.0'
+    implementation 'io.opentelemetry:opentelemetry-sdk:1.29.0'
+    implementation 'io.opentelemetry:opentelemetry-exporter-logging:1.29.0'
+    implementation 'io.opentelemetry:opentelemetry-exporter-otlp:1.29.0'
+    implementation 'io.opentelemetry:opentelemetry-semconv:1.29.0-alpha'
+    implementation 'com.google.code.gson:gson:2.10.1'
+    implementation group: 'io.github.microutils', name: 'kotlin-logging-jvm', version: '3.0.5'
+    implementation group: 'org.slf4j', name: 'slf4j-api', version: '1.7.36'
+    implementation group: 'ch.qos.logback', name: 'logback-classic', version: '1.3.11'
+    implementation group: 'ch.qos.logback.contrib', name: 'logback-json-classic', version: '0.1.5'
+    implementation group: 'ch.qos.logback.contrib', name: 'logback-jackson', version: '0.1.5'
+
+    implementation 'com.azure:azure-messaging-servicebus:7.15.0'
+    implementation 'com.microsoft.azure:azure-servicebus:3.6.7'
+    implementation 'com.azure:azure-identity:1.8.0'
+
+    implementation 'org.danilopianini:khttp:1.3.1'
+    implementation group: 'com.google.code.gson', name: 'gson', version: '2.8.9'
+
+    implementation 'org.owasp.encoder:encoder:1.2.3'
+    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.3.1'
+
+    agent "io.opentelemetry.javaagent:opentelemetry-javaagent:1.29.0"
+
+    testImplementation("org.mockito.kotlin:mockito-kotlin:4.0.0")
+    testImplementation "org.testng:testng:7.4.0"
+    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
+    testImplementation "org.mockito:mockito-inline:3.11.2"
+    testImplementation "io.mockk:mockk:1.13.9"
+
+
+}
+test {
+    useTestNG()
+    testLogging {
+        events "passed", "skipped", "failed"
+    }
+    //Change this to "true" if we want to execute unit tests
+    systemProperty("isTestEnvironment", "false")
+
+    // Set the test classpath, if required
 }
 
 jib {

pstatus-report-sink-ktor/src/main/kotlin/gov/cdc/ocio/processingstatusapi/Application.kt

@@ -1,5 +1,6 @@
 package gov.cdc.ocio.processingstatusapi
 
+import gov.cdc.ocio.processingstatusapi.cosmos.CosmosDeadLetterRepository
 import gov.cdc.ocio.processingstatusapi.cosmos.CosmosRepository
 import gov.cdc.ocio.processingstatusapi.plugins.configureRouting
 import gov.cdc.ocio.processingstatusapi.plugins.serviceBusModule
@@ -9,28 +10,40 @@ import io.ktor.server.netty.*
 import org.koin.core.KoinApplication
 import org.koin.dsl.module
 import org.koin.ktor.plugin.Koin
-import org.koin.mp.KoinPlatform.getKoin
 
+
+/**
+ * Load the environment configuration values
+ * Instantiate a singleton CosmosDatabase container instance
+ * @param environment ApplicationEnvironment
+ */
 fun KoinApplication.loadKoinModules(environment: ApplicationEnvironment): KoinApplication {
     val cosmosModule = module {
         val uri = environment.config.property("azure.cosmos_db.client.endpoint").getString()
         val authKey = environment.config.property("azure.cosmos_db.client.key").getString()
-        single { CosmosRepository(uri, authKey, "Reports", "/uploadId") }
+        single(createdAtStart = true) { CosmosRepository(uri, authKey, "Reports", "/uploadId") }
+        single(createdAtStart = true) { CosmosDeadLetterRepository(uri, authKey, "Reports-DeadLetter", "/uploadId") }
     }
+
     return modules(listOf(cosmosModule))
 }
 
+/**
+ * The main function
+ *  @param args Array<string>
+ */
 fun main(args: Array<String>) {
     embeddedServer(Netty, commandLineEnvironment(args)).start(wait = true)
 }
 
+/**
+ * The main application module which always runs and loads other modules
+ */
 fun Application.module() {
     configureRouting()
     serviceBusModule()
     install(Koin) {
         loadKoinModules(environment)
     }
 
-    // Preload the koin module so the CosmosDB client is already initialized on the first call
-    getKoin().get<CosmosRepository>()
 }