Fix json decoding empty object streams (#689)

* fix empty object stream decoding

* add a gen test

* move into the jvm only test folder

zio-schema-json/jvm/src/test/scala-2/zio/schema/codec/JsonCodecJVMSpec.scala

@@ -0,0 +1,67 @@
+package zio.schema.codec
+
+import java.nio.charset.StandardCharsets
+
+import zio.json.JsonStreamDelimiter
+import zio.schema.Schema
+import zio.schema.codec.JsonCodec.JsonEncoder.charSequenceToByteChunk
+import zio.schema.codec.JsonCodecSpec.AllOptionalFields
+import zio.stream.{ ZPipeline, ZStream }
+import zio.test.Assertion.{ equalTo, isRight }
+import zio.test.TestAspect.timeout
+import zio.test.{ Gen, Spec, TestAspect, TestEnvironment, ZIOSpecDefault, assertZIO, check }
+import zio.{ Chunk, durationInt }
+
+object JsonCodecJVMSpec extends ZIOSpecDefault {
+
+  def spec: Spec[TestEnvironment, Any] =
+    suite("JsonCodec JVM Spec")(
+      decoderSuite
+    ) @@ TestAspect.jvmOnly @@ timeout(180.seconds)
+
+  private val decoderSuite = suite("decoding")(
+    suite("case class")(
+      test("case class with empty option field is decoded by stream") {
+        val names     = Gen.option(Gen.elements("John", "Jane", "Jermaine", "Jasmine"))
+        val age       = Gen.option(Gen.int(1, 99))
+        val person    = names.zipWith(age)(AllOptionalFields(_, _, None))
+        val delimiter = Gen.elements(JsonStreamDelimiter.Array, JsonStreamDelimiter.Newline)
+        val codec     = JsonCodec.jsonEncoder(AllOptionalFields.schema)
+
+        check(Gen.chunkOfBounded(0, 3)(person), delimiter) {
+          (people, delim) =>
+            val indent        = if (delim == JsonStreamDelimiter.Array) Some(1) else None
+            val encodedPeople = people.map(p => codec.encodeJson(p, indent))
+            val encoded = delim match {
+              case JsonStreamDelimiter.Array =>
+                encodedPeople.mkString("[", ",", "]")
+              case JsonStreamDelimiter.Newline =>
+                encodedPeople.mkString("", "\n", "\n")
+            }
+
+            assertDecodesJsonStream(
+              AllOptionalFields.schema,
+              people,
+              charSequenceToByteChunk(encoded),
+              delim
+            )
+        }
+      }
+    )
+  )
+
+  private def assertDecodesJsonStream[A](
+    schema: Schema[A],
+    value: Chunk[A],
+    chunk: Chunk[Byte],
+    delimiter: JsonStreamDelimiter
+  ) = {
+    val result = ZStream
+      .fromChunk(chunk)
+      .via(ZPipeline.decodeCharsWith(StandardCharsets.UTF_8))
+      .via(JsonCodec.jsonDecoder(schema).decodeJsonPipeline(delimiter))
+      .runCollect
+      .either
+    assertZIO(result)(isRight(equalTo(value)))
+  }
+}

zio-schema-json/shared/src/main/scala/zio/schema/codec/JsonCodec.scala

@@ -3,13 +3,13 @@ package zio.schema.codec
 import java.nio.CharBuffer
 import java.nio.charset.StandardCharsets
 
-import scala.annotation.tailrec
+import scala.annotation.{ switch, tailrec }
 import scala.collection.immutable.ListMap
 
 import zio.json.JsonCodec._
 import zio.json.JsonDecoder.{ JsonError, UnsafeJson }
 import zio.json.ast.Json
-import zio.json.internal.{ Lexer, RecordingReader, RetractReader, StringMatrix, Write }
+import zio.json.internal.{ Lexer, RecordingReader, RetractReader, StringMatrix, WithRecordingReader, Write }
 import zio.json.{
   JsonCodec => ZJsonCodec,
   JsonDecoder => ZJsonDecoder,
@@ -78,9 +78,7 @@ object JsonCodec {
         )
 
       override def streamDecoder: ZPipeline[Any, DecodeError, Byte, A] =
-        ZPipeline.fromChannel(
-          ZPipeline.utfDecode.channel.mapError(cce => ReadError(Cause.fail(cce), cce.getMessage))
-        ) >>>
+        ZPipeline.utfDecode.mapError(cce => ReadError(Cause.fail(cce), cce.getMessage)) >>>
           ZPipeline.groupAdjacentBy[String, Unit](_ => ()) >>>
           ZPipeline.map[(Unit, NonEmptyChunk[String]), String] {
             case (_, fragments) => fragments.mkString
@@ -504,26 +502,33 @@ object JsonCodec {
       (_: List[JsonError], in: RetractReader) => {
         val c1 = in.nextNonWhitespace()
         val c2 = in.nextNonWhitespace()
-        if (c1 == '{' && c2 == '}') {
-          true
-        } else {
-          false
-        }
+        c1 == '{' && c2 == '}'
       }
 
     private[schema] def option[A](A: ZJsonDecoder[A]): ZJsonDecoder[Option[A]] =
       new ZJsonDecoder[Option[A]] { self =>
 
-        def unsafeDecode(trace: List[JsonError], in: RetractReader): Option[A] = {
-          val in2 = new zio.json.internal.WithRecordingReader(in, 2)
-          if (emptyObjectDecoder.unsafeDecode(trace, in2)) {
-            None
-          } else {
-            in2.retract()
-            in2.rewind()
-            zio.json.JsonDecoder.option(A).unsafeDecode(trace, in2)
+        private[this] val ull: Array[Char] = "ull".toCharArray
+
+        def unsafeDecode(trace: List[JsonError], in: RetractReader): Option[A] =
+          (in.nextNonWhitespace(): @switch) match {
+            case 'n' =>
+              Lexer.readChars(trace, in, ull, "null")
+              None
+            case '{' =>
+              // If we encounter a `{` it could either be a legitimate object or an empty object marker
+              in.retract()
+              val rr = new WithRecordingReader(in, 2)
+              if (emptyObjectDecoder.unsafeDecode(trace, rr)) {
+                None
+              } else {
+                rr.rewind()
+                Some(A.unsafeDecode(trace, rr))
+              }
+            case _ =>
+              in.retract()
+              Some(A.unsafeDecode(trace, in))
           }
-        }
 
         override def unsafeDecodeMissing(trace: List[JsonError]): Option[A] =
           None