Fix json decoding empty object streams

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