Use `Reactive Streams` to add stream support to `VertxFutureServerInterpreter` #2552

tdroxler · 2022-11-07T10:36:57Z

Following #2533 I gave it a try.

The change is backward compatible and people using VertxFutureServerInterpreter shouldn't have to change their endpoints, but if they want they can now use streaming.

Using Reactive Streams rather than Vertx Streams offers more flexibility. Any streaming library implementing reactivestreams.Publisher can be used.

So even zio, fs2, akka-streams etc can be used to handles streams in VertxFutureServerInterpreter.

I could successfully use this in this commit of my project

My endpoint type changed like this:

- type BaseEndpoint[I, O] = Endpoint[Unit, I, ApiError[_ <: StatusCode], O, Any]
+ type BaseEndpoint[I, O] = Endpoint[Unit, I, ApiError[_ <: StatusCode], O, ReactiveStreams]

Using a streamBody give me: BaseEndpoint[Address, Publisher[Buffer]]

In my code I then used akka-streams to create a Publisher, but I could have used anything else.

…erpreter` The change is backward compatible and people using `VertxFutureServerInterpreter` shouldn't have to change their endpoints, but if they want they can now use streaming. Using `Reactive Streams` rather than `Vertx Streams` offers more flexibility. Any streaming library implementing `reactivestreams.Publisher` can be used. So even `zio`, `fs2`, `akka-streams` etc can be used to handles streams in `VertxFutureServerInterpreter`.

Pask423

Hi
Thanks for PR could you please add server streaming tests for vertx.
Such line ++ new ServerStreamingTests(createServerTest, ReactiveStreams).tests() in VertxServerTest alongside adding ReactiveStreams type param in tests should do the job. If you will have any questions let me know.

tdroxler · 2022-11-07T16:59:35Z

Hi @Pask423,

I tried hard to make the tests pass, but there's something I don't understand in vertx :/ from what I get, pipeTo should be enough, but everything else I tried also failed.

Any chance to have a vertx expert to look at it?

I pushed a new commit with the tests.

Thx, cheers.

Pask423 · 2022-11-09T08:05:54Z

I think that it is unlikely, unfortunately - at least I do not know anyone who we could ask. I will try to take a look tomorrow or today. Maybe together we will come up with something.

tdroxler · 2022-11-09T08:19:34Z

I think that it is unlikely, unfortunately - at least I do not know anyone who we could ask. I will try to take a look tomorrow or today. Maybe together we will come up with something.

Thx, I'll also try to ask the vertx team directly

tdroxler · 2022-11-10T16:44:54Z

Sooooo I tried the all day to make the tests pass with the reactive-streams without success, when I do small test using only the conversion Publisher -> ReadStream and the opposite everything is fine with pipeTo. But then the all streams tests aren't passing, I have absolutely no idea why.

So I went with another proposition (latest commit) to use directly the vertx.streams.*, it offer a bit less freedom, but in my project I can easily use those ReadStream from a Publisher : https://github.com/alephium/explorer-backend/blob/6d9cde147b0ba9ec7cdc8bc9971140b6ad5cc680/app/src/main/scala/org/alephium/explorer/web/AddressServer.scala#L95-L98 , just need to subscribe my Publisher

What do you think?

Pask423 · 2022-11-14T10:18:03Z

Seems ok I came to similar conclusions some time ago - I think that this reactive streams lib is not intended to use in the way we want to use it here. However I would like you to change a few things in the code.

Pask423 · 2022-11-14T10:18:29Z

build.sbt

@@ -1258,7 +1258,8 @@ lazy val vertxServer: ProjectMatrix = (projectMatrix in file("server/vertx-serve
  .settings(
    name := "tapir-vertx-server",
    libraryDependencies ++= Seq(
-      "io.vertx" % "vertx-web" % Versions.vertx
+      "io.vertx" % "vertx-web" % Versions.vertx,
+      "io.vertx" % "vertx-reactive-streams" % Versions.vertx


At this point I think it is not used

Pask423 · 2022-11-14T10:19:16Z

server/vertx-server/src/main/scala/sttp/tapir/server/vertx/streams/ReadStreamCompatible.scala

+
+trait VertxStreams extends Streams[VertxStreams] {
+  override type BinaryStream = ReadStream[Buffer]
+  override type Pipe[A, B] = Processor[A, B]


I would go with ReadStream[A] => ReadStream[B] to not mix reactive streams and vertx streams

Pask423 · 2022-11-14T10:20:53Z

server/vertx-server/src/test/scala/sttp/tapir/server/vertx/VertxServerTest.scala

 import sttp.tapir.tests.{Test, TestSuite}

-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, Future}


remove this and .asInstanceOf - change VertxTestServerInterpreter to have VertxStreams instead of any. You will also have to change route contract in VertxTestServerBlockingInterpreter.

Pask423 · 2022-11-14T10:22:44Z

server/vertx-server/src/main/scala/sttp/tapir/server/vertx/streams/ReadStreamCompatible.scala

@@ -20,3 +21,10 @@ object ReadStreamCompatible {
    override def fromReadStream(s: ReadStream[Buffer]): Nothing = ???


remove incompatible - not needed anymore

adamw · 2022-11-14T12:36:12Z

Thanks! :)

adamw · 2022-11-14T12:37:36Z

It would also be great to add a Streaming subsection to https://tapir.softwaremill.com/en/latest/server/vertx.html like akka-http/http4s have

Pask423 self-assigned this Nov 7, 2022

Pask423 requested changes Nov 7, 2022

View reviewed changes

[FIXME] Add streaming tests to VertxServerTest

8a1cecb

Use VertxStreams instead of ReactiveStreams

8338e14

Pask423 requested changes Nov 14, 2022

View reviewed changes

Tackle PR comments

27991f4

Pask423 approved these changes Nov 14, 2022

View reviewed changes

adamw merged commit d3a462e into softwaremill:master Nov 14, 2022

This was referenced Nov 17, 2022

Feature/streaming report alephium/explorer-backend#389

Merged

Reactive Streams and vertx #2533

Closed

tdroxler mentioned this pull request Nov 24, 2022

Cannot cancel a stream #2593

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use `Reactive Streams` to add stream support to `VertxFutureServerInterpreter` #2552

Use `Reactive Streams` to add stream support to `VertxFutureServerInterpreter` #2552

tdroxler commented Nov 7, 2022

Pask423 left a comment •

edited

tdroxler commented Nov 7, 2022

Pask423 commented Nov 9, 2022 •

edited

tdroxler commented Nov 9, 2022

tdroxler commented Nov 10, 2022 •

edited

Pask423 commented Nov 14, 2022

Pask423 Nov 14, 2022 •

edited

tdroxler Nov 14, 2022

Pask423 Nov 14, 2022

tdroxler Nov 14, 2022

Pask423 Nov 14, 2022

tdroxler Nov 14, 2022

Pask423 Nov 14, 2022

tdroxler Nov 14, 2022

adamw commented Nov 14, 2022

adamw commented Nov 14, 2022

		@@ -20,3 +21,10 @@ object ReadStreamCompatible {
		override def fromReadStream(s: ReadStream[Buffer]): Nothing = ???

Use Reactive Streams to add stream support to VertxFutureServerInterpreter #2552

Use Reactive Streams to add stream support to VertxFutureServerInterpreter #2552

Conversation

tdroxler commented Nov 7, 2022

Pask423 left a comment • edited

Choose a reason for hiding this comment

tdroxler commented Nov 7, 2022

Pask423 commented Nov 9, 2022 • edited

tdroxler commented Nov 9, 2022

tdroxler commented Nov 10, 2022 • edited

Pask423 commented Nov 14, 2022

Pask423 Nov 14, 2022 • edited

Choose a reason for hiding this comment

tdroxler Nov 14, 2022

Choose a reason for hiding this comment

Pask423 Nov 14, 2022

Choose a reason for hiding this comment

tdroxler Nov 14, 2022

Choose a reason for hiding this comment

Pask423 Nov 14, 2022

Choose a reason for hiding this comment

tdroxler Nov 14, 2022

Choose a reason for hiding this comment

Pask423 Nov 14, 2022

Choose a reason for hiding this comment

tdroxler Nov 14, 2022

Choose a reason for hiding this comment

adamw commented Nov 14, 2022

adamw commented Nov 14, 2022

Use `Reactive Streams` to add stream support to `VertxFutureServerInterpreter` #2552

Use `Reactive Streams` to add stream support to `VertxFutureServerInterpreter` #2552

Pask423 left a comment •

edited

Pask423 commented Nov 9, 2022 •

edited

tdroxler commented Nov 10, 2022 •

edited

Pask423 Nov 14, 2022 •

edited