Force EOF after reading an object from stream

formats/json/commonMain/src/kotlinx/serialization/json/internal/lexer/AbstractJsonLexer.kt

@@ -158,7 +158,7 @@ internal abstract class AbstractJsonLexer {
     fun expectEof() {
         val nextToken = consumeNextToken()
         if (nextToken != TC_EOF)
-            fail("Expected EOF, but had ${source[currentPosition - 1]} instead")
+            fail("Expected EOF after parsing an object, but had ${source[currentPosition - 1]} instead")
     }
 
     /*

formats/json/jvmMain/src/kotlinx/serialization/json/JvmStreams.kt

@@ -1,3 +1,7 @@
+/*
+ * Copyright 2017-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
 package kotlinx.serialization.json
 
 import kotlinx.serialization.*
@@ -45,6 +49,9 @@ public inline fun <reified T> Json.encodeToStream(
 /**
  * Deserializes JSON from [stream] using UTF-8 encoding to a value of type [T] using [deserializer].
  *
+ * Note that this functions expects that exactly one object would be present in the stream
+ * and throws an exception if there are any dangling bytes after an object.
+ *
  * @throws [SerializationException] if the given JSON input cannot be deserialized to the value of type [T].
  * @throws [IOException] If an I/O error occurs and stream can't be read from.
  */
@@ -55,7 +62,9 @@ public fun <T> Json.decodeFromStream(
 ): T {
     val lexer = ReaderJsonLexer(stream)
     val input = StreamingJsonDecoder(this, WriteMode.OBJ, lexer, deserializer.descriptor)
-    return input.decodeSerializableValue(deserializer)
+    val result = input.decodeSerializableValue(deserializer)
+    lexer.expectEof()
+    return result
 }
 
 /**

formats/json/jvmTest/src/kotlinx/serialization/features/JsonStreamFlowTest.kt

@@ -1,9 +1,15 @@
+/*
+ * Copyright 2017-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
 package kotlinx.serialization.features
 
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.SerializationException
 import kotlinx.serialization.StringData
 import kotlinx.serialization.json.*
+import kotlinx.serialization.test.assertFailsWithMessage
 import org.junit.Ignore
 import org.junit.Test
 import java.io.*
@@ -18,14 +24,6 @@ class JsonStreamFlowTest {
         }
     }
 
-    suspend inline fun <reified T> InputStream.readToFlow(): Flow<T> {
-        return flow<T> {
-            while(available() != 0) {
-                emit(json.decodeFromStream(this@readToFlow))
-            }
-        }
-    }
-
     val inputString = """{"data":"a"}{"data":"b"}{"data":"c"}"""
     val inputList = listOf(StringData("a"), StringData("b"), StringData("c"))
 
@@ -42,14 +40,11 @@ class JsonStreamFlowTest {
     }
 
     @Test
-    @Ignore // todo: InputStream is consumed fully to buffer, looks like mechanism for multiple reading should be embedded in the framework itself
     fun testDecodeSeveralItems() {
         val ins = ByteArrayInputStream(inputString.encodeToByteArray())
-        val ml = mutableListOf<StringData>()
-        runBlocking {
-            ins.readToFlow<StringData>().toCollection(ml)
+        assertFailsWithMessage<SerializationException>("EOF") {
+            json.decodeFromStream<StringData>(ins)
         }
-        assertEquals(inputList, ml)
     }