Review fixes

Author: rossdanderson

util/src/main/kotlin/com/caplin/integration/datasourcex/util/flow/FlowMap.kt

@@ -95,6 +95,11 @@ sealed interface FlowMapStreamEvent<out K : Any, out V : Any> {
   /** Emitted for subsequent updates, containing only the delta ([event]). */
   @JvmInline
   value class EventUpdate<K : Any, V : Any>(val event: EntryEvent<K, V>) : FlowMapStreamEvent<K, V>
+
+  /** Emitted when the map is cleared. */
+  object Cleared : FlowMapStreamEvent<Nothing, Nothing> {
+    override fun toString(): String = "Cleared"
+  }
 }
 
 interface MapFlow<K : Any, V : Any> {
@@ -150,6 +155,11 @@ fun <K : Any, V : Any> Flow<FlowMapStreamEvent<K, V>>.runningFoldToMap(): Flow<M
         emit(map!!)
       }
 
+      is FlowMapStreamEvent.Cleared -> {
+        map = persistentMapOf()
+        emit(map!!)
+      }
+
       is FlowMapStreamEvent.EventUpdate -> {
         val currentMap = map ?: error("InitialState must be received before EventUpdate")
         when (val mapEvent = streamEvent.event) {
@@ -180,6 +190,7 @@ private class FlowMapImpl<K : Any, V : Any>(initialMap: PersistentMap<K, V>) :
   private data class FlowMapEvent<K : Any, V : Any>(
       val state: State<K, V>,
       val events: List<EntryEvent<K, V>>,
+      val isClear: Boolean = false,
   )
 
   private val state = MutableStateFlow(State(0L, initialMap))
@@ -281,9 +292,13 @@ private class FlowMapImpl<K : Any, V : Any>(initialMap: PersistentMap<K, V>) :
         emit(FlowMapStreamEvent.InitialState(flowMapEvent.state.map))
         first = false
       } else {
-        val events = flowMapEvent.events
-        for (event in events) {
-          emit(FlowMapStreamEvent.EventUpdate(event))
+        if (flowMapEvent.isClear) {
+          emit(FlowMapStreamEvent.Cleared)
+        } else {
+          val events = flowMapEvent.events
+          for (event in events) {
+            emit(FlowMapStreamEvent.EventUpdate(event))
+          }
         }
       }
     }
@@ -334,14 +349,20 @@ private class FlowMapImpl<K : Any, V : Any>(initialMap: PersistentMap<K, V>) :
   override fun containsKey(key: K): Boolean = state.value.map.containsKey(key)
 
   override fun putAll(from: Map<out K, V>) {
-    val (prev, next) = state.updateAndGetPrevAndNext { State(it.version + 1, it.map.putAll(from)) }
-
-    val events =
-        from.mapNotNull { (key, newValue) ->
-          val oldValue = prev.map[key]
-          if (newValue != oldValue) Upsert(key, oldValue, newValue) else null
+    val (prev, next) =
+        state.updateAndGetPrevAndNext {
+          val nextMap = it.map.putAll(from)
+          if (nextMap === it.map) it else State(it.version + 1, nextMap)
         }
-    if (events.isNotEmpty()) signal.tryEmit(FlowMapEvent(next, events))
+
+    if (prev != next) {
+      val events =
+          from.mapNotNull { (key, newValue) ->
+            val oldValue = prev.map[key]
+            if (newValue != oldValue) Upsert(key, oldValue, newValue) else null
+          }
+      if (events.isNotEmpty()) signal.tryEmit(FlowMapEvent(next, events))
+    }
   }
 
   override fun clear() {
@@ -350,7 +371,9 @@ private class FlowMapImpl<K : Any, V : Any>(initialMap: PersistentMap<K, V>) :
           if (it.map.isEmpty()) it else State(it.version + 1, it.map.clear())
         }
     if (prev.map.isNotEmpty())
-        signal.tryEmit(FlowMapEvent(next, prev.map.map { Removed(it.key, it.value) }))
+        signal.tryEmit(
+            FlowMapEvent(next, prev.map.map { Removed(it.key, it.value) }, isClear = true)
+        )
   }
 }
 

util/src/main/kotlin/com/caplin/integration/datasourcex/util/flow/SetEvent.kt

@@ -159,15 +159,19 @@ fun <V : Any, R> Flow<SetEvent<V>>.flatMapLatestAndMerge(
   val jobs = ConcurrentHashMap<V, Job>()
   collect { setEvent ->
     when (setEvent) {
-      is EntryEvent<V> -> {
+      is Insert<V> -> {
         jobs[setEvent.value]?.cancelAndJoin()
         jobs[setEvent.value] =
             entryEventTransformer(setEvent)
                 .onEach { send(it) }
-                .onCompletion { throwable -> if (throwable == null) jobs.remove(setEvent.value) }
+                .onCompletion { jobs.remove(setEvent.value) }
                 .launchIn(this@channelFlow)
       }
 
+      is Removed<V> -> {
+        jobs.remove(setEvent.value)?.cancelAndJoin()
+      }
+
       is Populated -> {}
     }
   }

util/src/main/kotlin/com/caplin/integration/datasourcex/util/flow/SimpleMapEvent.kt

@@ -113,6 +113,8 @@ fun <K : Any, V : Any> Flow<SimpleMapEvent<K, V>>.runningFoldToMap(
   var populated = false
   var map = persistentMapOf<K, V>()
 
+  if (emitPartials) emit(map)
+
   collect { mapEvent ->
     var emit = false
     when (mapEvent) {
@@ -130,8 +132,9 @@ fun <K : Any, V : Any> Flow<SimpleMapEvent<K, V>>.runningFoldToMap(
       }
 
       is Populated -> {
+        if (populated) error("Populated event already received")
         populated = true
-        if (!emitted || !emitPartials) emit = true
+        if (!emitted && !emitPartials) emit = true
       }
     }
     if (emit) {

util/src/main/kotlin/com/caplin/integration/datasourcex/util/flow/Timeout.kt

@@ -21,7 +21,7 @@ fun <T, R> Flow<T>.timeoutFirstOrDefault(millis: Long, default: () -> R): Flow<R
   val receiveChannel = produce { collect { send(it) } }
 
   select {
-    receiveChannel.onReceive { result -> send(result as R) }
+    receiveChannel.onReceiveCatching { result -> result.getOrNull()?.let { send(it as R) } }
     onTimeout(millis) { send(default()) }
   }
   receiveChannel.consumeEach { send(it as R) }

util/src/main/kotlin/com/caplin/integration/datasourcex/util/serialization/fory/FlowMapStreamEventSerializer.kt

@@ -12,6 +12,7 @@ internal class FlowMapStreamEventSerializer(fory: Fory, type: Class<FlowMapStrea
   private enum class Type {
     INITIAL_STATE,
     EVENT_UPDATE,
+    CLEARED,
   }
 
   override fun write(buffer: MemoryBuffer, value: FlowMapStreamEvent<*, *>) {
@@ -24,6 +25,9 @@ internal class FlowMapStreamEventSerializer(fory: Fory, type: Class<FlowMapStrea
         buffer.writeByte(Type.EVENT_UPDATE.ordinal.toByte())
         fory.writeRef(buffer, value.event)
       }
+      is FlowMapStreamEvent.Cleared -> {
+        buffer.writeByte(Type.CLEARED.ordinal.toByte())
+      }
     }
   }
 
@@ -37,6 +41,7 @@ internal class FlowMapStreamEventSerializer(fory: Fory, type: Class<FlowMapStrea
         val event = fory.readRef(buffer) as MapEvent.EntryEvent<Any, Any>
         FlowMapStreamEvent.EventUpdate(event)
       }
+      Type.CLEARED -> FlowMapStreamEvent.Cleared
     }
   }
 }

util/src/main/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson/FlowMapStreamEventDeserializer.kt

@@ -31,6 +31,7 @@ internal class FlowMapStreamEventDeserializer :
                 as MapEvent.EntryEvent<Any, Any>
         FlowMapStreamEvent.EventUpdate(event)
       }
+      "cleared" -> FlowMapStreamEvent.Cleared
       else -> throw JsonMappingException.from(p, "Unknown type: $type")
     }
   }

util/src/main/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson/FlowMapStreamEventSerializer.kt

@@ -24,6 +24,9 @@ internal class FlowMapStreamEventSerializer :
         gen.writeFieldName("event")
         provider.defaultSerializeValue(value.event, gen)
       }
+      is FlowMapStreamEvent.Cleared -> {
+        gen.writeStringField("type", "cleared")
+      }
     }
     gen.writeEndObject()
   }

util/src/test/kotlin/com/caplin/integration/datasourcex/util/flow/SimpleMapEventKtTest.kt

@@ -42,6 +42,7 @@ class SimpleMapEventKtTest :
             )
             .runningFoldToMap(emitPartials = true)
             .test {
+              awaitItem() shouldContainExactly emptyMap()
               awaitItem() shouldContainExactly mapOf("K" to "v1")
               awaitItem() shouldContainExactly mapOf("K" to "v2")
               awaitItem() shouldContainExactly mapOf("K" to "v2", "K2" to "v3")