refactoring EventSink

rodolfodpk · Jan 11, 2025 · 120c0f4 · 120c0f4
1 parent 677ed04
commit 120c0f4
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 7 deletions.
diff --git a/src/main/kotlin/crablet/query/CrabletQueryApi.kt b/src/main/kotlin/crablet/query/CrabletQueryApi.kt
@@ -12,11 +12,16 @@ data class SubscriptionSource(
 )
 
 sealed interface EventSink {
-    interface DefaultEventSink : EventSink {
-        fun handle(eventAsJson: List<JsonObject>): Future<Void>
+
+    interface SingleEventSink : EventSink {
+        fun handle(eventAsJson: JsonObject): Future<Void>
+    }
+
+    interface BatchEventSink : EventSink {
+        fun handle(eventAsJsonList: List<JsonObject>): Future<Void>
     }
 
-    interface PostgresEventSync : EventSink {
+    interface PostgresSingleEventSync : EventSink {
         fun handle(
             sqlConnection: SqlConnection,
             eventAsJson: JsonObject,

diff --git a/src/main/kotlin/crablet/query/impl/SubscriptionComponent.kt b/src/main/kotlin/crablet/query/impl/SubscriptionComponent.kt
@@ -30,15 +30,23 @@ class SubscriptionComponent(
                         rowSet.map { row -> row.toJson() }
                     }.flatMap { jsonList: List<JsonObject> ->
                         when (val eventSync = subscriptionConfig.eventSink) {
-                            is EventSink.PostgresEventSync -> {
+                            is EventSink.PostgresSingleEventSync -> {
                                 jsonList
                                     .fold(Future.succeededFuture<Void>()) { future, eventJson ->
                                         future.compose {
                                             eventSync.handle(tx, eventJson)
                                         }
                                     }.map { jsonList }
                             }
-                            is EventSink.DefaultEventSink -> eventSync.handle(jsonList)
+                            is EventSink.SingleEventSink -> {
+                                jsonList
+                                    .fold(Future.succeededFuture<Void>()) { future, eventJson ->
+                                        future.compose {
+                                            eventSync.handle(eventJson)
+                                        }
+                                    }.map { jsonList }
+                            }
+                            is EventSink.BatchEventSink -> eventSync.handle(jsonList)
                         }.map { jsonList }
                     }.compose { jsonList: List<JsonObject> ->
                         if (jsonList.isNotEmpty()) {

diff --git a/...t/query/AccountsPostgresEventProcessor.kt → ...y/AccountsPostgresSingleEventProcessor.kt b/...t/query/AccountsPostgresEventProcessor.kt → ...y/AccountsPostgresSingleEventProcessor.kt
@@ -5,7 +5,7 @@ import io.vertx.core.json.JsonObject
 import io.vertx.sqlclient.SqlConnection
 import io.vertx.sqlclient.Tuple
 
-class AccountsPostgresEventProcessor : EventSink.PostgresEventSync {
+class AccountsPostgresSingleEventProcessor : EventSink.PostgresSingleEventSync {
     override fun handle(
         sqlConnection: SqlConnection,
         eventAsJson: JsonObject,

diff --git a/src/test/kotlin/crablet/query/AccountsViewProjectionTest.kt b/src/test/kotlin/crablet/query/AccountsViewProjectionTest.kt
@@ -208,7 +208,7 @@ class AccountsViewProjectionTest : AbstractCrabletTest() {
 
             val subscriptionConfig = SubscriptionConfig(
                 source = source,
-                eventSink = AccountsPostgresEventProcessor(),
+                eventSink = AccountsPostgresSingleEventProcessor(),
                 callback = callback
             )