Add interop with Java Flow types #3102

BalmungSan · 2023-01-07T18:38:17Z

Adds support to converting Streams back and forth with Java 9+ Flow types.

Roadmap

Copy and paste the reactive-streams implementation inside core using Flow types instead.
Clean up the new versions.
Define an appropriate package for the new classes.
Add tests for the new classes.
Deprecate the reactive-streams module

core/jvm/src/main/scala/fs2/rx/package.scala

armanbilge · 2023-01-08T01:35:02Z

Looks like reactive streams has also published a testkit for the Java Flow counterparts.
https://www.reactive-streams.org/reactive-streams-tck-flow-1.0.2-javadoc/org/reactivestreams/tck/flow/package-summary.html

build.sbt

BalmungSan · 2023-01-08T22:12:12Z

Deprecate the reactive-streams module?

So, do we want to deprecate those or not?
And if so, what should be the message?

armanbilge · 2023-01-08T22:21:27Z

So, do we want to deprecate those or not?

Unfortunately I don't think we should "hard" deprecate it, because it leaves Java 8 users out in the cold.

Instead I propose a "soft" deprecation (like we did for Cats Future instances) that only appears in Scaladocs, not to the compiler.

The message can say that they will be removed in the next major version of FS2, and that the converters in this new package are the replacement.

armanbilge · 2023-01-08T22:41:13Z

Published as 3.4.0-59-7543784-SNAPSHOT for downstream testing.

Jasper-M · 2023-01-09T11:05:54Z

The message can say that they will be removed in the next major version of FS2, and that the converters in this new package are the replacement.

How would one interop with an API that is not updated to Java Flow in that scenario?

BalmungSan · 2023-01-09T12:29:19Z

@Jasper-M the same one interop with a Flow API in this moment, using the FlowAdapters provided by reactive-streams: https://www.reactive-streams.org/reactive-streams-flow-adapters-1.0.2-javadoc/org/reactivestreams/FlowAdapters.html
(just that it would be backwards in this case).

The rationale to prefer this approach is:

In theory, all APIs that expose reactive-streams should move to Flow types once they support Java 9.

This means that there will be a migratory period, while libraries move to adopt the new types in the JDK, however this period is expected to be short - due to the full semantic equivalence of the libraries, as well as the Reactive Streams <-> Flow adapter library as well as a TCK compatible directly with the JDK Flow types.
-- https://www.reactive-streams.org/

Right now, if you need to interop only with Flow types, you still need to include the reactive-streams library in your classpath. Whereas, with this new approach you won't be adding a new dependency.

armanbilge · 2023-01-09T15:00:49Z

core/jvm/src/main/scala/fs2/interop/flow/package.scala

+  def fromPublisher[F[_], A](
+      publisher: Publisher[A],
+      bufferSize: Int
+  )(implicit


Luis and I discussed this extensively offline. Unfortunately this current API is not usable as-is in the http4s-jdk-http-client, because it needs to run an effect in its Publisher. The workaround would be to unsafeRunSync() said effect.

However, noting that a Publisher[A] is just a @FunctionalInterface for Subscriber[A] => Unit which is a less ergonomic version of Subscriber[A] => F[Unit], I think we should change the signature to something like this:

def fromPublisher[F[_], A](bufferSize: Int)( subscribe: Subscription[A] => F[Unit] )(implicit F: Async[F]): Stream[F, A]

This also has a very nice parallel to existing APIs in FS2.

fs2/io/jvm/src/main/scala/fs2/io/ioplatform.scala

Lines 67 to 71 in 19e691f

def readOutputStream[F[_]: Async](

chunkSize: Int

)(

f: OutputStream => F[Unit]

): Stream[F, Byte] = {

I just pushed a couple of alternative APIs for this kind of use cases.

armanbilge · 2023-01-09T15:03:57Z

core/jvm/src/main/scala/fs2/interop/flow/package.scala

+  implicit final class PublisherOps[A](private val publisher: Publisher[A]) extends AnyVal {
+
+    /** Creates a [[Stream]] from an [[Publisher]].
+      *
+      * @param bufferSize setup the number of elements asked each time from the [[Publisher]].
+      *                   A high number can be useful is the publisher is triggering from IO,
+      *                   like requesting elements from a database.
+      *                   The publisher can use this `bufferSize` to query elements in batch.
+      *                   A high number will also lead to more elements in memory.
+      */
+    def toStream[F[_]](bufferSize: Int)(implicit F: Async[F]): Stream[F, A] =


I am a bit skeptical if we should expose this, for the same reasons we don't expose a toIO syntax for Future. Specifically, the signature does not ask for a raw Future, but a suspended one:

object IO { def fromFuture[A](fut: IO[Future[A]]): IO[A] }

Well, AFAIK a Publisher is just a description, you can create one without needing to run any side effects.

core/jvm/src/main/scala/fs2/interop/flow/StreamUnicastPublisher.scala

core/jvm/src/main/scala/fs2/interop/flow/package.scala

BalmungSan · 2023-01-19T04:36:14Z

core/jvm/src/main/scala/fs2/interop/flow/syntax.scala

+
+import java.util.concurrent.Flow.{Publisher, Subscriber}
+
+object syntax {


I moved the implicit classes to here from the package because:

To fix some annoying error with the doctest on 2.12

IMHO, is better not to import both the explicit methods and the syntax when doing import fs2.inteorp.flow._

To follow the standard of other libraries like cats and cats-effect

core/jvm/src/main/scala/fs2/interop/flow/syntax.scala

reactive-streams/src/main/scala/fs2/interop/reactivestreams/package.scala

Co-authored-by: Arman Bilge <armanbilge@gmail.com>

core/jvm/src/main/scala/fs2/interop/flow/package.scala

core/jvm/src/main/scala/fs2/interop/flow/StreamSubscriber.scala

armanbilge

Thanks for all your work, this is such a nice iteration on the reactive-streams module.

I'm going to publish one more snapshot. Update: 3.5.0-12-6d1e0b0-SNAPSHOT

armanbilge · 2023-01-20T02:27:49Z

31cf927 will be a follow-up PR :)

armanbilge · 2023-01-20T18:21:10Z

Looks like there is a defaultBufferSize(), should we make use of this anywhere?

BalmungSan · 2023-01-20T19:25:09Z

Looks like there is a defaultBufferSize(), should we make use of this anywhere?

Oh, I didn't see that one.
I actually thought of proposing a default for bufferSize but I wasn't sure which may be a good value.

IMHO, we should use this one on all public APIs

core/jvm/src/main/scala/fs2/interop/flow/package.scala

armanbilge reviewed Jan 8, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/rx/package.scala Outdated Show resolved Hide resolved

BalmungSan force-pushed the support-java-flow branch 2 times, most recently from 4d14048 to 3d43160 Compare January 8, 2023 02:01

BalmungSan commented Jan 8, 2023

View reviewed changes

build.sbt Outdated Show resolved Hide resolved

BalmungSan force-pushed the support-java-flow branch from 3d43160 to df588d3 Compare January 8, 2023 21:21

Add interop with Java Flow types

dcc58ca

BalmungSan force-pushed the support-java-flow branch from df588d3 to dcc58ca Compare January 8, 2023 21:30

BalmungSan changed the title ~~WIP: Add interop with Java Flow types~~ Add interop with Java Flow types Jan 8, 2023

BalmungSan marked this pull request as ready for review January 8, 2023 22:09

BalmungSan requested a review from armanbilge January 8, 2023 22:09

Add tests for Java Flow interop

7543784

BalmungSan force-pushed the support-java-flow branch from ec27e2b to 7543784 Compare January 8, 2023 22:18

Soft-deprecate the interop.reactivestreams package

3e2b9b1

armanbilge reviewed Jan 9, 2023

View reviewed changes

BalmungSan force-pushed the support-java-flow branch from 502046c to ec217bc Compare January 9, 2023 16:16

BalmungSan commented Jan 9, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/StreamUnicastPublisher.scala Outdated Show resolved Hide resolved

BalmungSan force-pushed the support-java-flow branch from ec217bc to d6651d8 Compare January 9, 2023 16:48

armanbilge reviewed Jan 9, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

BalmungSan force-pushed the support-java-flow branch 2 times, most recently from 6e547d8 to ef9304a Compare January 9, 2023 17:30

BalmungSan commented Jan 9, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

BalmungSan commented Jan 9, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

BalmungSan requested a review from armanbilge January 9, 2023 17:32

armanbilge reviewed Jan 9, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

Apply review comments

ba142a6

BalmungSan force-pushed the support-java-flow branch from 55841a0 to ba142a6 Compare January 19, 2023 04:06

BalmungSan commented Jan 19, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

BalmungSan commented Jan 19, 2023

View reviewed changes

Move extension methods to its own syntax object

29df7a8

BalmungSan force-pushed the support-java-flow branch from d08a4d9 to 29df7a8 Compare January 19, 2023 04:40

armanbilge reviewed Jan 19, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/syntax.scala Outdated Show resolved Hide resolved

BalmungSan force-pushed the support-java-flow branch from ee3f349 to c11a05a Compare January 19, 2023 04:57

Use the partially applied trick in fromPublisher

61061d9

BalmungSan force-pushed the support-java-flow branch from c11a05a to 61061d9 Compare January 19, 2023 04:59

armanbilge reviewed Jan 19, 2023

View reviewed changes

reactive-streams/src/main/scala/fs2/interop/reactivestreams/package.scala Outdated Show resolved Hide resolved

Fix Scaladoc link

10cc919

Co-authored-by: Arman Bilge <armanbilge@gmail.com>

armanbilge reviewed Jan 19, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Outdated Show resolved Hide resolved

core/jvm/src/main/scala/fs2/interop/flow/StreamSubscriber.scala Outdated Show resolved Hide resolved

Improve error handling

6d1e0b0

armanbilge approved these changes Jan 20, 2023

View reviewed changes

BalmungSan requested a review from armanbilge January 20, 2023 02:07

armanbilge force-pushed the support-java-flow branch from 31cf927 to 6d1e0b0 Compare January 20, 2023 02:21

armanbilge reviewed Jan 21, 2023

View reviewed changes

core/jvm/src/main/scala/fs2/interop/flow/package.scala Show resolved Hide resolved

Rename bufferSize as chunkSize

550f7cb

BalmungSan force-pushed the support-java-flow branch from 8835630 to 550f7cb Compare February 4, 2023 17:19

Merge branch 'main' into support-java-flow

9d4c22c

armanbilge approved these changes Feb 4, 2023

View reviewed changes

mpilquist merged commit a04e1be into typelevel:main Feb 7, 2023

BalmungSan deleted the support-java-flow branch February 7, 2023 03:35

BalmungSan mentioned this pull request Feb 7, 2023

Optimize the **Java** Flow interop #3130

Merged

3 tasks

rossabaker mentioned this pull request Feb 9, 2023

Flow type? http4s/http4s-netty#422

Open

peterneyens mentioned this pull request Oct 23, 2023

Use fs2 flow interop Banno/cosmos4s#254

Merged

	def readOutputStream[F[_]: Async](
	chunkSize: Int
	)(
	f: OutputStream => F[Unit]
	): Stream[F, Byte] = {


		import java.util.concurrent.Flow.{Publisher, Subscriber}

		object syntax {

Add interop with Java Flow types #3102

Add interop with Java Flow types #3102

Uh oh!

Conversation

BalmungSan commented Jan 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

armanbilge commented Jan 8, 2023

Uh oh!

Uh oh!

BalmungSan commented Jan 8, 2023

Uh oh!

armanbilge commented Jan 8, 2023

Uh oh!

armanbilge commented Jan 8, 2023

Uh oh!

Jasper-M commented Jan 9, 2023

Uh oh!

BalmungSan commented Jan 9, 2023

Uh oh!

armanbilge Jan 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

BalmungSan Jan 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

armanbilge Jan 9, 2023

Choose a reason for hiding this comment

Uh oh!

BalmungSan Jan 9, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

BalmungSan Jan 19, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

armanbilge left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

armanbilge commented Jan 20, 2023

Uh oh!

armanbilge commented Jan 20, 2023

Uh oh!

BalmungSan commented Jan 20, 2023

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

BalmungSan commented Jan 7, 2023 •

edited

Loading

armanbilge Jan 9, 2023 •

edited

Loading

BalmungSan Jan 9, 2023 •

edited

Loading

armanbilge left a comment •

edited

Loading