Endpoint API: Java ClassCastException when Using ZIO HTTP Streaming about zio-http

[guersam]

Something similar happens with RC6 for Server Sent Events. (scastie)

[guersam]

I think there's a room for improvement in stream body encoding.

For example, maybe SSE deserves a dedicated HttpContentCodec rather than the hidden Schema comparison.

[guersam]

I think I've spotted the cause:

1. Endpoint#outStream[A] summons an implicit HttpContentCodec (code)
2. New HttpContentCodec is created using the existing schema, with application/json as the first candidate (code)
3. The JSON codec is picked up
4. The encoder checks if the default media type is binary (code)
5. JSON is defined as a binary codec (code)
6. The encoder unsafely casts the stream as byte-emitting without properly encoding it (code)
7. NettyWriter crashes as Body.StreamBody assumes it contains a byte stream

[guersam]

application/json has been treated as binary since 2022, unlike the other JSON variants. I think it's just a mistake.

[guersam]

For SSE, I took a workaround with a custom codec that might be added to the upstream once there is a proper decoder implementation.

val binaryCodec: BinaryCodec[ServerSentEvent] =
  new BinaryCodec[ServerSentEvent] {
    override def encode(value: ServerSentEvent): Chunk[Byte] =
      Chunk.fromArray(value.encode.getBytes)

    override def streamEncoder: ZPipeline[Any, Nothing, ServerSentEvent, Byte] =
      ZPipeline.fromFunction(_.mapChunks(_.flatMap(encode)))

    override def decode(whole: Chunk[Byte]): Either[DecodeError, ServerSentEvent] =
      Left(DecodeError.UnsupportedSchema(ServerSentEvent.schema, "ServerSendEvent"))

    override def streamDecoder: ZPipeline[Any, DecodeError, Byte, ServerSentEvent] =
      ZPipeline.fromSink(ZSink.fail(DecodeError.UnsupportedSchema(ServerSentEvent.schema, "ServerSendEvent")))
  }

val httpContentCodec: HttpContentCodec[ServerSentEvent] =
  HttpContentCodec(
    ListMap(
      MediaType.text.`event-stream` -> binaryCodec
    ),
    ServerSentEvent.schema
  )

val httpCodec: HttpCodec[HttpCodecType.Content, ZStream[Any, Nothing, ServerSentEvent]] =
  HttpCodec.contentStream(httpContentCodec)