add logging

VEuPathDB · Jun 11, 2022 · 6625733 · 6625733
1 parent f2d00a5
commit 6625733
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 6 deletions.
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -9,7 +9,7 @@ plugins {
 }
 
 group = "org.veupathdb.lib"
-version = "1.1.1"
+version = "1.2.0"
 
 repositories {
   mavenCentral()
@@ -27,6 +27,8 @@ dependencies {
   implementation(kotlin("stdlib"))
   implementation(kotlin("stdlib-jdk8"))
 
+  implementation("org.slf4j:slf4j-api:1.7.36")
+
   // Jackson and modules (gotta catch em all)
   implementation("com.fasterxml.jackson.core:jackson-databind:2.13.3")
   implementation("com.fasterxml.jackson.datatype:jackson-datatype-json-org:2.13.3")

diff --git a/src/main/kotlin/org/veupathdb/lib/rabbit/jobs/QueueDispatcher.kt b/src/main/kotlin/org/veupathdb/lib/rabbit/jobs/QueueDispatcher.kt
@@ -2,6 +2,7 @@ package org.veupathdb.lib.rabbit.jobs
 
 import com.rabbitmq.client.CancelCallback
 import com.rabbitmq.client.DeliverCallback
+import org.slf4j.LoggerFactory
 import org.veupathdb.lib.rabbit.jobs.fn.ErrorHandler
 import org.veupathdb.lib.rabbit.jobs.fn.SuccessHandler
 import org.veupathdb.lib.rabbit.jobs.model.ErrorNotification
@@ -15,6 +16,9 @@ import org.veupathdb.lib.rabbit.jobs.serialization.Json
  * Job dispatcher.
  */
 class QueueDispatcher : QueueWrapper {
+
+  private val Log = LoggerFactory.getLogger(javaClass)
+
   private val errorHandlers = ErrorHandlers()
 
   private val successHandlers = SuccessHandlers()
@@ -29,6 +33,7 @@ class QueueDispatcher : QueueWrapper {
    * @param fn Success callback.
    */
   fun onSuccess(fn: SuccessHandler) {
+    Log.debug("registering success handler {}", fn)
     successHandlers.register(fn)
   }
 
@@ -38,6 +43,7 @@ class QueueDispatcher : QueueWrapper {
    * @param fn Error callback.
    */
   fun onError(fn: ErrorHandler) {
+    Log.debug("registering error handler {}", fn)
     errorHandlers.register(fn)
   }
 
@@ -47,6 +53,7 @@ class QueueDispatcher : QueueWrapper {
    * @param job Job definition.
    */
   fun dispatch(job: JobDispatch) {
+    Log.debug("dispatching job {}", job)
     withDispatchQueue { publish(dispatchQueueName, job) }
   }
 
@@ -56,8 +63,14 @@ class QueueDispatcher : QueueWrapper {
         errorQueueName,
         false,
         DeliverCallback { _, msg ->
+          Log.debug("handling error message {}", msg.envelope.deliveryTag)
           workers.execute {
-            errorHandlers.execute(ErrorNotification.fromJson(Json.from(msg.body)))
+            try {
+              errorHandlers.execute(ErrorNotification.fromJson(Json.from(msg.body)))
+            } finally {
+              Log.debug("acknowledging error message {}", msg.envelope.deliveryTag)
+              basicAck(msg.envelope.deliveryTag, false)
+            }
           }
         },
         CancelCallback { }
@@ -69,8 +82,14 @@ class QueueDispatcher : QueueWrapper {
         successQueueName,
         false,
         DeliverCallback { _, msg ->
+          Log.debug("handling success message {}", msg.envelope.deliveryTag)
           workers.execute {
-            successHandlers.execute(SuccessNotification.fromJson(Json.from(msg.body)))
+            try {
+              successHandlers.execute(SuccessNotification.fromJson(Json.from(msg.body)))
+            } finally {
+              Log.debug("acknowledging success message {}", msg.envelope.deliveryTag)
+              basicAck(msg.envelope.deliveryTag, false)
+            }
           }
         },
         CancelCallback { }

diff --git a/src/main/kotlin/org/veupathdb/lib/rabbit/jobs/QueueWorker.kt b/src/main/kotlin/org/veupathdb/lib/rabbit/jobs/QueueWorker.kt
@@ -2,6 +2,7 @@ package org.veupathdb.lib.rabbit.jobs
 
 import com.rabbitmq.client.CancelCallback
 import com.rabbitmq.client.DeliverCallback
+import org.slf4j.LoggerFactory
 import org.veupathdb.lib.rabbit.jobs.fn.JobHandler
 import org.veupathdb.lib.rabbit.jobs.model.ErrorNotification
 import org.veupathdb.lib.rabbit.jobs.model.JobDispatch
@@ -13,6 +14,9 @@ import org.veupathdb.lib.rabbit.jobs.serialization.Json
  * Job executor end of the job queue.
  */
 class QueueWorker : QueueWrapper {
+
+  private val Log = LoggerFactory.getLogger(javaClass)
+
   private val handlers = JobHandlers()
 
   /**
@@ -37,6 +41,7 @@ class QueueWorker : QueueWrapper {
    * @param fn Job request callback.
    */
   fun onJob(fn: JobHandler) {
+    Log.debug("registering job handler {}", fn)
     handlers.register(fn)
   }
 
@@ -47,6 +52,7 @@ class QueueWorker : QueueWrapper {
    * @param err Error notification to send.
    */
   fun sendError(err: ErrorNotification) {
+    Log.debug("sending error notification {}", err)
     withErrorQueue { publish(errorQueueName, err) }
   }
 
@@ -57,6 +63,7 @@ class QueueWorker : QueueWrapper {
    * @param msg Success notification to send.
    */
   fun sendSuccess(msg: SuccessNotification) {
+    Log.debug("sending success notification {}", msg)
     withSuccessQueue { publish(successQueueName, msg) }
   }
 
@@ -67,10 +74,16 @@ class QueueWorker : QueueWrapper {
     withDispatchQueue {
       basicConsume(
         dispatchQueueName,
-        true,
+        false,
         DeliverCallback { _, msg ->
+          Log.debug("handling job message {}", msg.envelope.deliveryTag)
           workers.execute {
-            handlers.execute(JobDispatch.fromJson(Json.from(msg.body)))
+            try {
+              handlers.execute(JobDispatch.fromJson(Json.from(msg.body)))
+            } finally {
+              Log.debug("acknowledging job message {}", msg.envelope.deliveryTag)
+              basicAck(msg.envelope.deliveryTag, false)
+            }
           }
         },
         CancelCallback { }

diff --git a/test/server/build.gradle.kts b/test/server/build.gradle.kts
@@ -27,7 +27,7 @@ dependencies {
   implementation(kotlin("stdlib-jdk8"))
   implementation(rootProject)
   implementation("org.veupathdb.lib:hash-id:1.0.2")
-  implementation("com.fasterxml.jackson.core:jackson-databind:2.13.0")
+  implementation("com.fasterxml.jackson.core:jackson-databind:2.13.3")
 }
 
 kotlin {