BOC can track message consumption

airbytehq · Sep 16, 2024 · 50234f4 · 50234f4
1 parent 10cea0b
commit 50234f4
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/...k/bulk/core/base/src/testFixtures/kotlin/io/airbyte/cdk/output/BufferingOutputConsumer.kt b/...k/bulk/core/base/src/testFixtures/kotlin/io/airbyte/cdk/output/BufferingOutputConsumer.kt
@@ -34,6 +34,7 @@ class BufferingOutputConsumer(
     private val catalogs = mutableListOf<AirbyteCatalog>()
     private val traces = mutableListOf<AirbyteTraceMessage>()
     private val messages = mutableListOf<AirbyteMessage>()
+    private var messagesIndex: Int = 0
 
     var callback: (AirbyteMessage) -> Unit = {}
         set(value) {
@@ -79,4 +80,11 @@ class BufferingOutputConsumer(
     fun traces(): List<AirbyteTraceMessage> = synchronized(this) { listOf(*traces.toTypedArray()) }
 
     fun messages(): List<AirbyteMessage> = synchronized(this) { listOf(*messages.toTypedArray()) }
+
+    fun newMessages(): List<AirbyteMessage> =
+        synchronized(this) {
+            val newMessages = messages.subList(messagesIndex, messages.size)
+            messagesIndex = messages.size
+            newMessages
+        }
 }