Merge pull request #105 from AxonIQ/feature/add-total-number-for-pagination-support

stefanmirkovic · web-flow · commit 3e4665b103f2 · 2025-09-17T21:32:51.000+02:00
Total number for pagination support
diff --git a/console-framework-client-api/pom.xml b/console-framework-client-api/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.10.1-SNAPSHOT</version>
+        <version>1.10.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/Routes.kt b/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/Routes.kt
@@ -57,8 +57,10 @@ object Routes {
             const val SEQUENCE_SIZE = "dlq-query-dead-letter-sequence-size"
 
             const val DELETE_SEQUENCE = "dlq-command-delete-sequence"
+            const val DELETE_ALL_SEQUENCES = "dlq-command-delete-all-sequences"
             const val DELETE_LETTER = "dlq-command-delete-letter"
             const val PROCESS = "dlq-command-process"
+            const val PROCESS_ALL_SEQUENCES = "dlq-command-process-all-sequences"
         }
     }
 
diff --git a/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/deadLetterApi.kt b/console-framework-client-api/src/main/java/io/axoniq/console/framework/api/deadLetterApi.kt
@@ -31,7 +31,8 @@ data class DeadLetter(
 )
 
 data class DeadLetterResponse(
-    val sequences: List<List<DeadLetter>>
+    val sequences: List<List<DeadLetter>>,
+    val totalCount: Long = -1
 )
 
 data class DeadLetterRequest(
@@ -57,6 +58,15 @@ data class DeadLetterSingleDeleteRequest(
     val messageIdentifier: String,
 )
 
+data class ProcessAllDeadLetterSequencesRequest(
+    val processingGroup: String,
+    val maxMessages: Int = 10
+)
+
+data class DeleteAllDeadLetterSequencesRequest(
+        val processingGroup: String
+)
+
 data class DeadLetterProcessRequest(
     val processingGroup: String,
     val messageIdentifier: String
diff --git a/console-framework-client-spring-boot-starter/pom.xml b/console-framework-client-spring-boot-starter/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.10.1-SNAPSHOT</version>
+        <version>1.10.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client/pom.xml b/console-framework-client/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.axoniq.console</groupId>
         <artifactId>console-framework-client-parent</artifactId>
-        <version>1.10.1-SNAPSHOT</version>
+        <version>1.10.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/DeadLetterManager.kt b/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/DeadLetterManager.kt
@@ -28,6 +28,7 @@ import java.util.concurrent.Callable
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.TimeUnit
 import io.axoniq.console.framework.api.DeadLetter as ApiDeadLetter
+import io.axoniq.console.framework.api.DeadLetterResponse
 
 private const val LETTER_PAYLOAD_SIZE_LIMIT = 1024
 private const val MASKED = "<MASKED>"
@@ -46,20 +47,21 @@ class DeadLetterManager(
         offset: Int = 0,
         size: Int = 25,
         maxSequenceLetters: Int = 10
-    ): List<List<ApiDeadLetter>> {
+    ): DeadLetterResponse {
         if (dlqMode == AxoniqConsoleDlqMode.NONE) {
-            return emptyList()
+            return DeadLetterResponse(emptyList(), 0)
         }
-        return dlqFor(processingGroup)
-            .deadLetters()
-            .drop(offset)
-            .take(size)
-            .map { sequence ->
-                sequence
-                    .asIterable()
-                    .take(maxSequenceLetters)
-                    .map { toDeadLetter(it, processingGroup) }
-            }
+        val queue = dlqFor(processingGroup)
+        val sequences = queue
+                .deadLetters()
+                .drop(offset)
+                .take(size)
+                .map { sequence ->
+                    sequence.asIterable()
+                            .take(maxSequenceLetters)
+                            .map { toDeadLetter(it, processingGroup) }
+                }
+        return DeadLetterResponse(sequences, queue.amountOfSequences())
     }
 
     private fun toDeadLetter(letter: DeadLetter<out EventMessage<*>>, processingGroup: String) =
@@ -169,6 +171,38 @@ class DeadLetterManager(
         }).get(60, TimeUnit.SECONDS)
     }
 
+    fun processAll(
+            processingGroup: String,
+            maxMessages: Int? = null,
+            timeoutSeconds: Long = 600
+    ): Int {
+        return executor.submit(Callable {
+            val processor = letterProcessorFor(processingGroup)
+            var processedCount = 0
+
+            // Process all messages or up to maxMessages limit
+            while ((maxMessages == null || processedCount < maxMessages)) {
+                val processed = processor.process { true }
+                if (!processed) break // No more messages to process
+                processedCount++
+            }
+
+            processedCount
+        }).get(timeoutSeconds, TimeUnit.SECONDS)
+    }
+
+    fun deleteAll(
+            processingGroup: String,
+            timeoutSeconds: Long = 600
+    ): Int {
+        return executor.submit(Callable {
+            val dlq = dlqFor(processingGroup)
+            val totalCount = dlq.size().toInt()
+            dlq.clear()
+            totalCount
+        }).get(timeoutSeconds, TimeUnit.SECONDS)
+    }
+
     private fun String.hashIfNeeded(): String {
         return if (dlqMode == AxoniqConsoleDlqMode.MASKED) {
             DigestUtils.sha256Hex(this)
diff --git a/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/RSocketDlqResponder.kt b/console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/RSocketDlqResponder.kt
@@ -58,11 +58,21 @@ open class RSocketDlqResponder(
             DeadLetterProcessRequest::class.java,
             this::handleProcessCommand
         )
+        registrar.registerHandlerWithPayload(
+                Routes.ProcessingGroup.DeadLetter.PROCESS_ALL_SEQUENCES,
+                ProcessAllDeadLetterSequencesRequest::class.java,
+                this::handleProcessAllSequencesCommand
+        )
+        registrar.registerHandlerWithPayload(
+                Routes.ProcessingGroup.DeadLetter.DELETE_ALL_SEQUENCES,
+                DeleteAllDeadLetterSequencesRequest::class.java,
+                this::handleDeleteAllSequencesCommand
+        )
     }
 
     private fun handleDeadLetterQuery(request: DeadLetterRequest): DeadLetterResponse {
         logger.debug("Handling AxonIQ Console DEAD_LETTERS query for request [{}]", request)
-        return DeadLetterResponse(deadLetterManager.deadLetters(request.processingGroup, request.offset, request.size))
+        return deadLetterManager.deadLetters(request.processingGroup, request.offset, request.size)
     }
 
     private fun handleSequenceSizeQuery(request: DeadLetterSequenceSize): Long {
@@ -93,4 +103,15 @@ open class RSocketDlqResponder(
         logger.debug("Handling AxonIQ Console DEAD LETTERS query for processing group [{}]", request.processingGroup)
         return deadLetterManager.process(request.processingGroup, request.messageIdentifier)
     }
+
+    private fun handleProcessAllSequencesCommand(request: ProcessAllDeadLetterSequencesRequest): Int {
+        logger.debug("Handling AxonIQ Console PROCESS_ALL_DEAD_LETTER_SEQUENCES commands for processing group [{}]", request.processingGroup)
+        return deadLetterManager.processAll(request.processingGroup, request.maxMessages)
+    }
+
+    private fun handleDeleteAllSequencesCommand(request: DeleteAllDeadLetterSequencesRequest): Int {
+        logger.debug("Handling AxonIQ Console DELETE_ALL_DEAD_LETTER_SEQUENCES commands for processing group [{}]", request.processingGroup)
+        return deadLetterManager.deleteAll(request.processingGroup)
+    }
+
 }
diff --git a/pom.xml b/pom.xml
@@ -22,7 +22,7 @@
 
     <groupId>io.axoniq.console</groupId>
     <artifactId>console-framework-client-parent</artifactId>
-    <version>1.10.1-SNAPSHOT</version>
+    <version>1.10.2-SNAPSHOT</version>
 
     <modules>
         <module>console-framework-client-api</module>

Original file line number	Diff line number	Diff line change
`@@ -57,8 +57,10 @@ object Routes {`
`57`	`57`	`const val SEQUENCE_SIZE = "dlq-query-dead-letter-sequence-size"`
`58`	`58`
`59`	`59`	`const val DELETE_SEQUENCE = "dlq-command-delete-sequence"`
	`60`	`+ const val DELETE_ALL_SEQUENCES = "dlq-command-delete-all-sequences"`
`60`	`61`	`const val DELETE_LETTER = "dlq-command-delete-letter"`
`61`	`62`	`const val PROCESS = "dlq-command-process"`
	`63`	`+ const val PROCESS_ALL_SEQUENCES = "dlq-command-process-all-sequences"`
`62`	`64`	`}`
`63`	`65`	`}`
`64`	`66`