Netty streaming for Cats Effect #3017

kciesielski · 2023-07-07T10:41:53Z

This PR enhances Netty Cats Effect server with streaming.
From now on, the public API is:

  NettyCatsServer
    .io()
    .use { server =>
      server
        .port(port)
        .host(host)
        .addEndpoint(serverEndpoint) // <<< ServerEndpoint[Fs2Streams[F]]
        .start()
    }

Implementation notes:

I used HttpStreamsServerHandler from netty-reactive-streams-http. Adding it to the pipeline allows using a response of type DefaultStreamedHttpResponse, which wraps a reactive Publisher[HttpContent].
However, this handler doesn't work together with HttpObjectAggregator, so I had to split the pipeline into streaming/non-streaming variants.
Without HttpObjectAggregator we no longer can send ChunkedInput, but this isn't a problem, we can just send files as streams as well.
A bigger problem is handling requests. Without HttpObjectAggregator we need to handle request which have a body as StreamedHttpRequest, which have a .subscribe method. Using fs2, I create streams which read data from this .subscribe interface and materialize it into byte arrays as raw values. This may introduce performance overhead.
Empty requests fall through the HttpStreamsServerHandler as instances of a regular FullHttpRequest.

adamw · 2023-07-17T18:01:56Z

examples/src/main/scala/sttp/tapir/examples/streaming/StreamingNettyFs2Server.scala

+    .io()
+    .use { server =>
+
+      val effect: IO[NettyCatsServerBinding[IO]] = server


[minor] effect -> sth descriptive, e.g. startServer?

adamw · 2023-07-17T18:02:09Z

doc/server/netty.md

-* `NettyCatsServer().addEndpoints` to expose `F`-based server endpoints, where `F` is any cats-effect supported effect.
-* `NettyZioServer().addEndpoints` to expose `ZIO`-based server endpoints, where `R` represents ZIO requirements supported effect.
+- `NettyFutureServer().addEndpoints` to expose `Future`-based server endpoints.
+- `NettyCatsServer().addEndpoints` to expose `F`-based server endpoints, where `F` is any cats-effect supported effect. [Streaming](../endpoint/streaming.md) request and response body is supported with fs2.


[minor] shouldn't this be "bodies"?

adamw · 2023-07-17T18:09:10Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala

    if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler())
    ()
  }

+  def defaultWithStreaming: NettyConfig = default.copy(initPipeline = cfg => streamingPipeline(cfg)(_, _))


maybe we should be more explicit here and rename default to defaultNoStreaming, same with defaultInitPipeliene, so that there's no ambiguity

adamw · 2023-07-17T18:24:54Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

+          new DefaultStreamedHttpResponse(req.protocolVersion(), HttpResponseStatus.valueOf(serverResponse.code.code), publisher)
+
+        res.setHeadersFrom(serverResponse)
+        res.handleContentLengthAndChunkedHeaders(None)


will this work if the content-length is explicitly set by the user in headers?

I'll check if there's a test for this, but my first thought is we don't want users to set content-length when streaming. It's either transfer-encoding: chunked (stream) or explicit content-length. (4.4 https://greenbytes.de/tech/webdav/rfc2616.html#rfc.section.4.4)

well the user might know what the length of the streaming body is going to be upfront. There's even a test for it:

tapir/server/tests/src/main/scala/sttp/tapir/server/tests/ServerStreamingTests.scala

Line 37 in c3f45cb

.contentLength(penPineapple.length.toLong)

, though it doesn't check if the response is chunked or not. Maybe some implementations don't allow setting conent-lenght on streaming bodies 🤔 This should really be under a test flag, not an if in the test

If the specification always requires chunked response to be without the content-length header, then this if is IMO correct, this shouldn't depend on implementation, right?

How we produce the data (streaming / non-streaming) doesn't have to directly translate to chunked/non-chunked encoding. We can have content with known length, and just produce it lazily

adamw · 2023-07-18T07:35:35Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

-      val req = request.retain()
+      request match {
+        case full: FullHttpRequest =>
+          val req = full.retain()


is the only difference between those branches that we call retain/release in one case, and not in the other? If so, maybe we can extract the common part to a method

adamw · 2023-07-18T07:36:31Z

...ty-server/cats/src/main/scala/sttp/tapir/server/netty/internal/NettyCatsToResponseBody.scala

+      case RawBodyType.InputStreamBody =>
+        val stream = inputStreamToFs2(() => v)
+        (ctx: ChannelHandlerContext) =>
+          new NettyResponseContent.ReactivePublisherNettyResponseContent(ctx.newPromise(), fs2StreamToPublisher(stream))


[minor] other classes don't have the prefix

I mean other usages of the classes :)

adamw · 2023-07-18T07:39:13Z

...ty-server/cats/src/main/scala/sttp/tapir/server/netty/internal/NettyCatsToResponseBody.scala

+
+  override def fromRawValue[R](v: R, headers: HasHeaders, format: CodecFormat, bodyType: RawBodyType[R]): NettyResponse = {
+    bodyType match {
+      case RawBodyType.StringBody(charset) =>


I think most branches are the same as in NettyToResponseBody? Maybe we can handle the different cases here, and for the raw body types that are the same as in NettyToResponseBody, delegate to that impl?

adamw · 2023-07-18T07:46:47Z

Looks good :)

kciesielski · 2023-07-18T08:00:08Z

There's one more thing I need to add here, I'm working on it now. Since we're no longer using HttpObjectAggregator for the Cats Netty server, the check for maxContentLength has to be reimplemented in another way. It's fairly easy to add it for regular ByteBuf response body, but a bit tricky for streaming body.

adamw · 2023-07-18T08:03:09Z

@kciesielski this also might be part of another task (or sub-task) here: #2971 - I don't think there are any tests for this, for example

kciesielski · 2023-07-18T08:04:01Z

@adamw I just wrote one, for the regular byte buf body. But for the streaming, I was about to suggest the same - to make it another task :)

- This needs to go to a separate PR

adamw · 2023-07-19T07:42:51Z

Awesome! :)

I'd still add a task to polish the max content length, though - maybe we can somehow reuse what the user provides in the exception handler, to create the response? Or create another interceptor? Also, we should make sure that if a limit exists, we never read into memory more than the limit specifies.

kciesielski added 5 commits July 15, 2023 03:00

wip

17f4647

wip

2f35d51

Streaming for Netty Cats server

c04918b

Adjust addEndpoint to accept streaming endpoints

305253b

Mention streaming in docs

24e64e7

kciesielski force-pushed the feature/netty-streaming branch from 3dac45d to 24e64e7 Compare July 15, 2023 01:00

kciesielski changed the title ~~[wip] netty streaming~~ Netty streaming for Cats Effect Jul 15, 2023

kciesielski added 6 commits July 15, 2023 03:09

Minor cleanup

94b00ee

Additional cleanup in tests

739d6f1

Make netty-reactive-streams version a val

dcfff16

Remove unused val

0e9f04f

Add an example

47a1097

More cleanup

70863aa

kciesielski marked this pull request as ready for review July 17, 2023 06:49

kciesielski requested a review from adamw July 17, 2023 06:49

adamw reviewed Jul 17, 2023

View reviewed changes

adamw reviewed Jul 18, 2023

View reviewed changes

kciesielski added 4 commits July 18, 2023 10:11

Handle maxContentLength

aa92c09

Remove maxContentLength handling for streams

72b6d53

- This needs to go to a separate PR

Refactoring after review

cec1cca

Let Netty Handle content-length and close handlers for streaming

2712d6a

kciesielski added 3 commits July 18, 2023 13:32

Flag for checking setting content-length header in streaming responses

0064b12

Update overlooked references to NettyConfig.default

fd8383f

Always expect content-length in streaming response if set by user

7dcdc6d

kciesielski requested a review from adamw July 19, 2023 06:43

adamw merged commit f5c73cf into master Jul 19, 2023

mergify bot deleted the feature/netty-streaming branch July 19, 2023 07:43

kciesielski mentioned this pull request Jul 24, 2023

Netty streaming for ZIO #3062

Merged

kamilkloch mentioned this pull request Jul 24, 2023

Netty websockets for Cats Effect #3066

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Netty streaming for Cats Effect #3017

Netty streaming for Cats Effect #3017

kciesielski commented Jul 7, 2023 •

edited

Loading

adamw Jul 17, 2023

adamw Jul 17, 2023

adamw Jul 17, 2023

adamw Jul 17, 2023

kciesielski Jul 18, 2023

adamw Jul 18, 2023

kciesielski Jul 18, 2023

adamw Jul 18, 2023

adamw Jul 18, 2023

adamw Jul 18, 2023

adamw Jul 18, 2023

adamw Jul 18, 2023

adamw commented Jul 18, 2023

kciesielski commented Jul 18, 2023

adamw commented Jul 18, 2023

kciesielski commented Jul 18, 2023

adamw commented Jul 19, 2023

Netty streaming for Cats Effect #3017

Netty streaming for Cats Effect #3017

Conversation

kciesielski commented Jul 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adamw commented Jul 18, 2023

kciesielski commented Jul 18, 2023

adamw commented Jul 18, 2023

kciesielski commented Jul 18, 2023

adamw commented Jul 19, 2023

kciesielski commented Jul 7, 2023 •

edited

Loading