fixes an issue with json encoding post bodies

Author: halfhp

src/main/scala/com/example/scalaspringexperiment/util/CirceJsonDecoder.scala

@@ -1,39 +1,66 @@
 package com.example.scalaspringexperiment.util
 
 import io.circe.Json
-import io.circe.parser.decode
+import io.circe.parser
 import org.reactivestreams.Publisher
 import org.springframework.core.ResolvableType
 import org.springframework.core.codec.Decoder
-import org.springframework.core.io.buffer.DataBuffer
-import org.springframework.core.io.buffer.DataBufferUtils
-import org.springframework.http.{MediaType, ResponseEntity}
+import org.springframework.core.io.buffer.{DataBuffer, DataBufferUtils}
+import org.springframework.http.MediaType
 import org.springframework.util.MimeType
-import reactor.core.publisher.Mono
-import reactor.core.publisher.Flux
+import reactor.core.publisher.{Flux, Mono}
 
 import java.nio.charset.StandardCharsets
 import java.util
-import java.lang.reflect.Type
 
 class CirceJsonDecoder extends Decoder[io.circe.Json] {
 
-  override def canDecode(elementType: ResolvableType, mimeType: MimeType): Boolean = {
+  override def canDecode(
+    elementType: ResolvableType,
+    mimeType: MimeType
+  ): Boolean = {
     mimeType == null || mimeType.isCompatibleWith(MediaType.APPLICATION_JSON)
   }
 
-  override def decode(input: Publisher[DataBuffer], elementType: ResolvableType, mimeType: MimeType, hints: util.Map[String, AnyRef]): Flux[io.circe.Json] = {
-    Flux.from(input).flatMap { buffer =>
-      val jsonStr = StandardCharsets.UTF_8.decode(buffer.asByteBuffer()).toString
-      DataBufferUtils.release(buffer)
-      io.circe.parser.decode[io.circe.Json](jsonStr) match {
-        case Right(json) => Flux.just(json)
-        case Left(err)   => Flux.error(new RuntimeException(s"JSON decoding error: ${err.getMessage}"))
+
+  /**
+   * TODO: there may be performance issues with how this method handles incoming JSON bodies;
+   * particularly that it waits for the entire body to be read before decoding it.
+   * need to investigate if this is the best approach to use with WebFlux
+   *
+   * @param input
+   * @param elementType
+   * @param mimeType
+   * @param hints
+   * @return
+   */
+  override def decode(
+    input: Publisher[DataBuffer],
+    elementType: ResolvableType,
+    mimeType: MimeType,
+    hints: util.Map[String, AnyRef]
+  ): Flux[Json] = {
+    DataBufferUtils.join(Flux.from(input)).flatMap { joinedBuffer =>
+      val inputStream = joinedBuffer.asInputStream()
+      try {
+        val jsonStr = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8)
+        parser.decode[Json](jsonStr) match {
+          case Right(json) => Mono.just(json)
+          case Left(err) => Mono.error(err)
+        }
+      } finally {
+        DataBufferUtils.release(joinedBuffer)
+        inputStream.close()
       }
-    }
+    }.flux()
   }
 
-  override def decodeToMono(input: Publisher[DataBuffer], elementType: ResolvableType, mimeType: MimeType, hints: util.Map[String, AnyRef]): Mono[io.circe.Json] = {
+  override def decodeToMono(
+    input: Publisher[DataBuffer],
+    elementType: ResolvableType,
+    mimeType: MimeType,
+    hints: util.Map[String, AnyRef]
+  ): Mono[io.circe.Json] = {
     decode(input, elementType, mimeType, hints).single()
   }
 

src/main/scala/com/example/scalaspringexperiment/util/CirceJsonEncoder.scala

@@ -1,6 +1,8 @@
 package com.example.scalaspringexperiment.util
 
+
 import io.circe.Json
+import org.reactivestreams.Publisher
 import org.springframework.core.ResolvableType
 import org.springframework.core.codec.Encoder
 import org.springframework.core.io.buffer.{DataBuffer, DataBufferFactory}
@@ -10,22 +12,36 @@ import reactor.core.publisher.Flux
 
 import java.nio.charset.StandardCharsets
 import java.util
-import org.reactivestreams.Publisher
 
 class CirceJsonEncoder extends Encoder[io.circe.Json] {
 
-  override def canEncode(elementType: ResolvableType, mimeType: MimeType): Boolean =
+  override def canEncode(
+    elementType: ResolvableType,
+    mimeType: MimeType
+  ): Boolean =
     mimeType == null || mimeType.isCompatibleWith(MediaType.APPLICATION_JSON)
 
-  override def encode(inputStream: Publisher[_ <: io.circe.Json], bufferFactory: DataBufferFactory, elementType: ResolvableType, mimeType: MimeType, hints: util.Map[String, AnyRef]): Flux[DataBuffer] = {
+  override def encode(
+    inputStream: Publisher[? <: io.circe.Json],
+    bufferFactory: DataBufferFactory,
+    elementType: ResolvableType,
+    mimeType: MimeType,
+    hints: util.Map[String, AnyRef]
+  ): Flux[DataBuffer] = {
     Flux.from(inputStream).map { json =>
       val bytes = json.noSpaces.getBytes(StandardCharsets.UTF_8)
       val buffer = bufferFactory.wrap(bytes)
       buffer
     }
   }
 
-  override def encodeValue(value: io.circe.Json, bufferFactory: DataBufferFactory, elementType: ResolvableType, mimeType: MimeType, hints: util.Map[String, AnyRef]): DataBuffer = {
+  override def encodeValue(
+    value: io.circe.Json,
+    bufferFactory: DataBufferFactory,
+    elementType: ResolvableType,
+    mimeType: MimeType,
+    hints: util.Map[String, AnyRef]
+  ): DataBuffer = {
     val bytes = value.noSpaces.getBytes(StandardCharsets.UTF_8)
     bufferFactory.wrap(bytes)
   }