Drop native java serialization support, add fory and jackson serializers, and add benchmarks

Author: rossdanderson

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

@@ -90,7 +90,8 @@ interface FlowMap<K : Any, V : Any> : MapFlow<K, V>, Map<K, V> {
 sealed interface FlowMapStreamEvent<out K : Any, out V : Any> {
   /** Emitted on initial collection, containing the entire initial [map] state. */
   @JvmInline
-  value class InitialState<K : Any, V : Any>(val map: PersistentMap<K, V>) : FlowMapStreamEvent<K, V>
+  value class InitialState<K : Any, V : Any>(val map: PersistentMap<K, V>) :
+      FlowMapStreamEvent<K, V>
 
   /** Emitted for subsequent updates, containing only the delta ([event]). */
   @JvmInline

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

@@ -6,22 +6,27 @@ import org.apache.fory.Fory
 import org.apache.fory.serializer.collection.MapSerializer
 
 /**
- * Registers serializers for [PersistentMap] and other internal types with the provided [Fory] instance.
+ * Registers serializers for [PersistentMap] and other internal types with the provided [Fory]
+ * instance.
  */
 fun Fory.registerDataSourceSerializers(): Fory = apply {
   registerSerializer(PersistentMap::class.java, PersistentMapSerializer::class.java)
   // Kotlin immutable collections implementations often have internal names
   val hashMapClass =
-      runCatching { Class.forName("kotlinx.collections.immutable.implementations.immutableMap.PersistentHashMap") }.getOrNull()
+      runCatching {
+            Class.forName(
+                "kotlinx.collections.immutable.implementations.immutableMap.PersistentHashMap"
+            )
+          }
+          .getOrNull()
   if (hashMapClass != null) {
     registerSerializer(hashMapClass, PersistentMapSerializer::class.java)
   }
 }
 
-/**
- * A Fory [MapSerializer] for [PersistentMap].
- */
-class PersistentMapSerializer(fory: Fory, type: Class<PersistentMap<*, *>>) : MapSerializer<PersistentMap<*, *>>(fory, type) {
+/** A Fory [MapSerializer] for [PersistentMap]. */
+class PersistentMapSerializer(fory: Fory, type: Class<PersistentMap<*, *>>) :
+    MapSerializer<PersistentMap<*, *>>(fory, type) {
 
   @Suppress("UNCHECKED_CAST")
   override fun onMapRead(map: Map<*, *>): PersistentMap<*, *> {

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

@@ -25,8 +25,7 @@ import kotlinx.coroutines.flow.transformWhile
 sealed interface ValueOrCompletion<out T> {
 
   @Suppress("UNCHECKED_CAST")
-  suspend fun <R> map(block: suspend (T) -> R): ValueOrCompletion<R> =
-      this as ValueOrCompletion<R>
+  suspend fun <R> map(block: suspend (T) -> R): ValueOrCompletion<R> = this as ValueOrCompletion<R>
 
   class Value<out T>(val value: T) : ValueOrCompletion<T> {
     operator fun component1(): T = value

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

@@ -14,10 +14,8 @@ import org.apache.fory.config.Language
 
 class FlowMapSerializationTest :
     FunSpec({
-
       context("Jackson Serialization") {
-        val mapper: ObjectMapper = jacksonObjectMapper()
-            .registerDataSourceModule()
+        val mapper: ObjectMapper = jacksonObjectMapper().registerDataSourceModule()
 
         test("serialize and deserialize InitialState") {
           val map = mutableFlowMapOf("1" to "A", "2" to "B")
@@ -26,7 +24,10 @@ class FlowMapSerializationTest :
 
           val json = mapper.writeValueAsString(initialState)
           val deserialized: FlowMapStreamEvent<String, String> =
-              mapper.readValue(json, object : TypeReference<FlowMapStreamEvent<String, String>>() {})
+              mapper.readValue(
+                  json,
+                  object : TypeReference<FlowMapStreamEvent<String, String>>() {},
+              )
 
           deserialized.shouldBeInstanceOf<FlowMapStreamEvent.InitialState<String, String>>()
           deserialized.map shouldContainExactly mapOf("1" to "A", "2" to "B")
@@ -38,7 +39,10 @@ class FlowMapSerializationTest :
 
           val json = mapper.writeValueAsString(event)
           val deserialized: FlowMapStreamEvent<String, String> =
-              mapper.readValue(json, object : TypeReference<FlowMapStreamEvent<String, String>>() {})
+              mapper.readValue(
+                  json,
+                  object : TypeReference<FlowMapStreamEvent<String, String>>() {},
+              )
 
           deserialized.shouldBeInstanceOf<FlowMapStreamEvent.EventUpdate<String, String>>()
           deserialized.event.shouldBeInstanceOf<Upsert<String, String>>()
@@ -61,7 +65,10 @@ class FlowMapSerializationTest :
 
       context("Apache Fory Serialization") {
         val fory =
-            Fory.builder().withLanguage(Language.JAVA).requireClassRegistration(false).build()
+            Fory.builder()
+                .withLanguage(Language.JAVA)
+                .requireClassRegistration(false)
+                .build()
                 .registerDataSourceSerializers()
 
         test("serialize and deserialize InitialState (raw PersistentMap)") {