Avoid another copy on write

Author: rossdanderson

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

@@ -6,6 +6,7 @@ import java.time.Duration
 import java.util.concurrent.TimeoutException
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.channels.onSuccess
 import kotlinx.coroutines.channels.produce
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
@@ -21,7 +22,7 @@ fun <T, R> Flow<T>.timeoutFirstOrDefault(millis: Long, default: () -> R): Flow<R
   val receiveChannel = produce { collect { send(it) } }
 
   select {
-    receiveChannel.onReceiveCatching { result -> result.getOrNull()?.let { send(it as R) } }
+    receiveChannel.onReceiveCatching { result -> result.onSuccess { send(it as R) } }
     onTimeout(millis) { send(default()) }
   }
   receiveChannel.consumeEach { send(it as R) }
@@ -45,7 +46,7 @@ fun <T, R> Flow<T>.timeoutFirstOrDefault(millis: Long, default: R): Flow<R> =
  * If the upstream emits no first event within [duration] it will emit the event [default] followed
  * by all later emissions from the upstream.
  */
-fun <T, R : Any?> Flow<T>.timeoutFirstOrDefault(duration: Duration, default: R): Flow<R> =
+fun <T, R> Flow<T>.timeoutFirstOrDefault(duration: Duration, default: R): Flow<R> =
     timeoutFirstOrDefault(duration.toMillis(), default)
 
 /**

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

@@ -8,24 +8,27 @@ import org.apache.fory.serializer.collection.MapSerializer
 
 /** A Fory [MapSerializer] for [PersistentMap] (Hash implementation). */
 internal class PersistentHashMapSerializer(fory: Fory, type: Class<PersistentMap<*, *>>) :
-    MapSerializer<PersistentMap<*, *>>(fory, type) {
+    MapSerializer<PersistentMap<*, *>>(fory, type, true) {
 
-  override fun write(buffer: MemoryBuffer, value: PersistentMap<*, *>) {
-    fory.writeRef(buffer, HashMap(value))
+  override fun newMap(buffer: MemoryBuffer): MutableMap<*, *> {
+    val numElements = buffer.readVarUint32Small7()
+    setNumElements(numElements)
+    val map = HashMap<Any?, Any?>(numElements)
+    refResolver.reference(map)
+    return map
   }
 
-  @Suppress("UNCHECKED_CAST")
-  override fun read(buffer: MemoryBuffer): PersistentMap<*, *> {
-    val map = fory.readRef(buffer) as Map<Any, Any>
-    return map.toPersistentHashMap()
+  override fun newMap(map: Map<*, *>): MutableMap<*, *> {
+    return HashMap<Any?, Any?>(map.size)
   }
 
-  override fun newMap(buffer: MemoryBuffer): MutableMap<*, *> {
-    return HashMap<Any, Any>()
+  @Suppress("UNCHECKED_CAST")
+  override fun onMapRead(map: Map<*, *>): PersistentMap<*, *> {
+    return (map as Map<Any, Any>).toPersistentHashMap()
   }
 
   @Suppress("UNCHECKED_CAST")
-  override fun onMapRead(map: Map<*, *>): PersistentMap<*, *> {
+  override fun onMapCopy(map: Map<*, *>): PersistentMap<*, *> {
     return (map as Map<Any, Any>).toPersistentHashMap()
   }
 }

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

@@ -8,24 +8,27 @@ import org.apache.fory.serializer.collection.CollectionSerializer
 
 /** A Fory [CollectionSerializer] for [PersistentSet] (Hash implementation). */
 internal class PersistentHashSetSerializer(fory: Fory, type: Class<PersistentSet<*>>) :
-    CollectionSerializer<PersistentSet<*>>(fory, type) {
+    CollectionSerializer<PersistentSet<*>>(fory, type, true) {
 
-  override fun write(buffer: MemoryBuffer, value: PersistentSet<*>) {
-    fory.writeRef(buffer, HashSet(value))
+  override fun newCollection(buffer: MemoryBuffer): MutableCollection<*> {
+    val numElements = buffer.readVarUint32Small7()
+    setNumElements(numElements)
+    val set = HashSet<Any?>(numElements)
+    refResolver.reference(set)
+    return set
   }
 
-  @Suppress("UNCHECKED_CAST")
-  override fun read(buffer: MemoryBuffer): PersistentSet<*> {
-    val collection = fory.readRef(buffer) as Collection<Any>
-    return collection.toPersistentHashSet()
+  override fun newCollection(collection: Collection<*>): MutableCollection<*> {
+    return HashSet<Any?>(collection.size)
   }
 
-  override fun newCollection(buffer: MemoryBuffer): MutableCollection<*> {
-    return HashSet<Any>()
+  @Suppress("UNCHECKED_CAST")
+  override fun onCollectionRead(collection: Collection<*>): PersistentSet<*> {
+    return (collection as Collection<Any>).toPersistentHashSet()
   }
 
   @Suppress("UNCHECKED_CAST")
-  override fun onCollectionRead(collection: Collection<*>): PersistentSet<*> {
+  fun onCollectionCopy(collection: Collection<*>): PersistentSet<*> {
     return (collection as Collection<Any>).toPersistentHashSet()
   }
 }

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

@@ -8,25 +8,27 @@ import org.apache.fory.serializer.collection.MapSerializer
 
 /** A Fory [MapSerializer] for [PersistentMap] (Ordered implementation). */
 internal class PersistentOrderedMapSerializer(fory: Fory, type: Class<PersistentMap<*, *>>) :
-    MapSerializer<PersistentMap<*, *>>(fory, type) {
+    MapSerializer<PersistentMap<*, *>>(fory, type, true) {
 
-  override fun write(buffer: MemoryBuffer, value: PersistentMap<*, *>) {
-    fory.writeRef(buffer, LinkedHashMap(value))
+  override fun newMap(buffer: MemoryBuffer): MutableMap<*, *> {
+    val numElements = buffer.readVarUint32Small7()
+    setNumElements(numElements)
+    val map = LinkedHashMap<Any?, Any?>(numElements)
+    refResolver.reference(map)
+    return map
   }
 
-  @Suppress("UNCHECKED_CAST")
-  override fun read(buffer: MemoryBuffer): PersistentMap<*, *> {
-    val map = fory.readRef(buffer) as Map<Any, Any>
-    return map.toPersistentMap()
+  override fun newMap(map: Map<*, *>): MutableMap<*, *> {
+    return LinkedHashMap<Any?, Any?>(map.size)
   }
 
-  override fun newMap(buffer: MemoryBuffer): MutableMap<*, *> {
-    return LinkedHashMap<Any, Any>()
+  @Suppress("UNCHECKED_CAST")
+  override fun onMapRead(map: Map<*, *>): PersistentMap<*, *> {
+    return (map as Map<Any, Any>).toPersistentMap()
   }
 
   @Suppress("UNCHECKED_CAST")
-  override fun onMapRead(map: Map<*, *>): PersistentMap<*, *> {
-    // toPersistentMap() on a LinkedHashMap preserves order in kotlinx-collections-immutable
+  override fun onMapCopy(map: Map<*, *>): PersistentMap<*, *> {
     return (map as Map<Any, Any>).toPersistentMap()
   }
 }

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

@@ -8,24 +8,27 @@ import org.apache.fory.serializer.collection.CollectionSerializer
 
 /** A Fory [CollectionSerializer] for [PersistentSet] (Ordered implementation). */
 internal class PersistentOrderedSetSerializer(fory: Fory, type: Class<PersistentSet<*>>) :
-    CollectionSerializer<PersistentSet<*>>(fory, type) {
+    CollectionSerializer<PersistentSet<*>>(fory, type, true) {
 
-  override fun write(buffer: MemoryBuffer, value: PersistentSet<*>) {
-    fory.writeRef(buffer, LinkedHashSet(value))
+  override fun newCollection(buffer: MemoryBuffer): MutableCollection<*> {
+    val numElements = buffer.readVarUint32Small7()
+    setNumElements(numElements)
+    val set = LinkedHashSet<Any?>(numElements)
+    refResolver.reference(set)
+    return set
   }
 
-  @Suppress("UNCHECKED_CAST")
-  override fun read(buffer: MemoryBuffer): PersistentSet<*> {
-    val collection = fory.readRef(buffer) as Collection<Any>
-    return collection.toPersistentSet()
+  override fun newCollection(collection: Collection<*>): MutableCollection<*> {
+    return LinkedHashSet<Any?>(collection.size)
   }
 
-  override fun newCollection(buffer: MemoryBuffer): MutableCollection<*> {
-    return LinkedHashSet<Any>()
+  @Suppress("UNCHECKED_CAST")
+  override fun onCollectionRead(collection: Collection<*>): PersistentSet<*> {
+    return (collection as Collection<Any>).toPersistentSet()
   }
 
   @Suppress("UNCHECKED_CAST")
-  override fun onCollectionRead(collection: Collection<*>): PersistentSet<*> {
+  fun onCollectionCopy(collection: Collection<*>): PersistentSet<*> {
     return (collection as Collection<Any>).toPersistentSet()
   }
 }