fix(util): round-trip null Value in ValueOrCompletion serialization (#47)

A materialized Flow<T?> that emits null produces Value(null), but none
of the serializers could round-trip it:

- Fory force-cast the read ref to non-null Any (NullPointerException).
- The Jackson 2/3 deserializers conflated an absent value field with a
  present-but-null one, throwing "Missing value field" on "value": null.

Drop the Fory non-null cast, and in both Jackson deserializers throw only
when the field is genuinely absent, treating a present null node as null.

Adds a null-Value round-trip test to each serialization spec.

Author: rossdanderson

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

@@ -38,7 +38,7 @@ internal class ValueOrCompletionSerializer(
   override fun read(readContext: ReadContext): ValueOrCompletion<*> {
     return when (Type.entries[readContext.readByte().toInt()]) {
       Type.VALUE -> {
-        val value = readContext.readRef() as Any
+        val value = readContext.readRef()
         ValueOrCompletion.Value(value)
       }
       Type.COMPLETION -> {

util/src/main/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson2/ValueOrCompletionDeserializer.kt

@@ -16,9 +16,10 @@ internal class ValueOrCompletionDeserializer :
             ?: throw JsonMappingException.from(p, "Missing type field for ValueOrCompletion")
     return when (type) {
       "value" -> {
-        val value =
-            node.get("value")?.let { p.codec.treeToValue(it, Any::class.java) }
+        val valueNode =
+            node.get("value")
                 ?: throw JsonMappingException.from(p, "Missing value field for value type")
+        val value = if (valueNode.isNull) null else p.codec.treeToValue(valueNode, Any::class.java)
         ValueOrCompletion.Value(value)
       }
       "completion" -> {

util/src/main/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson3/ValueOrCompletionDeserializer.kt

@@ -15,9 +15,10 @@ internal class ValueOrCompletionDeserializer :
             ?: throw DatabindException.from(p, "Missing type field for ValueOrCompletion")
     return when (type) {
       "value" -> {
-        val value =
-            node.get("value")?.let { ctxt.readTreeAsValue(it, Any::class.java) }
+        val valueNode =
+            node.get("value")
                 ?: throw DatabindException.from(p, "Missing value field for value type")
+        val value = if (valueNode.isNull) null else ctxt.readTreeAsValue(valueNode, Any::class.java)
         ValueOrCompletion.Value(value)
       }
       "completion" -> {

util/src/test/kotlin/com/caplin/integration/datasourcex/util/serialization/fory/ValueOrCompletionSerializationTest.kt

@@ -50,6 +50,13 @@ class ValueOrCompletionSerializationTest :
             deserialized shouldBe event
           }
 
+          test("Value (null)") {
+            val event = ValueOrCompletion.Value<String?>(null)
+            val bytes = fory.serialize(event)
+            val deserialized = fory.deserialize(bytes)
+            deserialized shouldBe event
+          }
+
           test("Completion") {
             val event = ValueOrCompletion.Completion(null)
             val bytes = fory.serialize(event)
@@ -92,7 +99,7 @@ class ValueOrCompletionSerializationTest :
               val bytes = foryNoType.serialize(event)
               val deserialized = foryNoType.deserialize(bytes) as ValueOrCompletion.Completion
               deserialized.throwable.shouldBeInstanceOf<RuntimeException>()
-              deserialized.throwable?.message shouldBe "aah"
+              deserialized.throwable.message shouldBe "aah"
             }
           }
         },

util/src/test/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson2/ValueOrCompletionSerializationTest.kt

@@ -20,6 +20,14 @@ class ValueOrCompletionSerializationTest :
         deserialized shouldBe event
       }
 
+      test("Value (null)") {
+        val event: ValueOrCompletion<String?> = ValueOrCompletion.Value(null)
+        val json = mapper.writeValueAsString(event)
+        val deserialized =
+            mapper.readValue(json, object : TypeReference<ValueOrCompletion<String?>>() {})
+        deserialized shouldBe event
+      }
+
       test("Completion (null)") {
         val event: ValueOrCompletion<String> = ValueOrCompletion.Completion(null)
         val json = mapper.writeValueAsString(event)

util/src/test/kotlin/com/caplin/integration/datasourcex/util/serialization/jackson3/ValueOrCompletionSerializationTest.kt

@@ -19,6 +19,14 @@ class ValueOrCompletionSerializationTest :
         deserialized shouldBe event
       }
 
+      test("Value (null)") {
+        val event: ValueOrCompletion<String?> = ValueOrCompletion.Value(null)
+        val json = mapper.writeValueAsString(event)
+        val deserialized =
+            mapper.readValue(json, object : TypeReference<ValueOrCompletion<String?>>() {})
+        deserialized shouldBe event
+      }
+
       test("Completion (null)") {
         val event: ValueOrCompletion<String> = ValueOrCompletion.Completion(null)
         val json = mapper.writeValueAsString(event)