Reactive-stream/Flow Interop #896

HollandDM · 2024-12-04T11:46:52Z

Originally, this was created just to resolve #764, but since #881 was requested, I took sometime to create a dedicated PR for it.
This took both ZIO's and Fs2's approaches as the inspiration, and passes the reactive-stream-tck testcases. I also add some test cases derived from Fs2's codebase. Hopefully it covers enough.
With this, solving #764 will be more trivial.

HollandDM · 2024-12-04T14:15:55Z

...reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscription.scala

+        tag: Tag[Emit[Chunk[V]]],
+        frame: Frame,
+        safepoint: Safepoint
+    ): Fiber[Nothing, StreamFinishState] < (IO & Ctx) =


Basically, we need to fork this into a new fiber, which will run the stream until completion. Maybe Poll can help us eliminate the need to fork, which results in eliminating boundary

Yeah, I can't see why forking is necessary. Wouldn't Poll.run(stream.emit)(poll).map(_._2) directly work?

Other libs like ZIO and Fs2 also have to fork. AFAIK, it's mainly because draining stream is a "runForever" task until interrupted or stream is done. Without forking it will block other thing from being done in the current thread.
I'll try to figure out if I can eliminate it

HollandDM · 2024-12-06T11:23:25Z

...reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscription.scala

+
+    private[reactivestreams] inline def subscribe: Unit < IO = IO(subscriber.onSubscribe(this))
+
+    private[reactivestreams] def poll: StreamFinishState < (Async & Poll[Chunk[V]]) =


Poll is great, is make the logic much cleaner then the previous attemp

calvinlfer · 2024-12-06T16:43:51Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/package.scala

+package object reactivestreams:
+    def fromPublisher[T](
+        publisher: Publisher[T],
+        bufferSize: Int


I had a question regarding this, do we wait for the entire buffer to be filled up before emitting them as a Chunk on the Stream? Is it possible to have some kind of choice with regards to size & time?

The reason I ask this is because on an infinite stream (let's say we're integrating with a reactive streams kafka library like Helidon or Vertx Kafka), if this buffer size > number of messages on the topic unconsumed then the messages will be held in the buffer until number of messages on the topic unconsumed >= buffer size.

I observed this limitation with the Helidon Kafka client when I was integrating with the FS2 Streams Flow interop where if you used a low buffer size, you can overcome this limitation at the cost of performance since chunk-size = 1 has much worse performance

Current implementation tries to emit the value to stream as soon as it receives one. Buffer size was used mostly as a params of subscription’s ’request’.

I did notice this behavior of fs2 stream and I’m also not a fan of it. So I took the Zio approach. But maybe we can have some kind of “strategy” or different implementations so that users can choose between them. WDYT

That would be excellent! Amazing work 🙇

fwbrasil · 2024-12-09T23:48:55Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/package.scala

+import kyo.kernel.Boundary
+import org.reactivestreams.FlowAdapters
+
+package object flow:


What do you think about exposing these operations as Stream extensions in the kyo package instead? I'd be nice to keep a companion object as well since people might want to import it in isolation:

package kyo object StreamReactiveStreamsExtensions: extension [V, S](self: Stream[V, S]): // instance methods extension (self: Stream.type): // companion methods // export to the kyo package export StreamReactiveStreamsExtensions.*

sounds good, I'' update this

now that I though about it, moving this into kyo package would polute it with reactive-streams dependencies. I don't think that's a good ideal, right?

It'd still be limited to the specific module. kyo-reactive-streams needs to be in the classpath for its extension methods to be available

I understand what you mean now, will update!

fwbrasil · 2024-12-10T00:19:01Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/package.scala

+        using
+        Boundary[Ctx, IO],
+        Frame,
+        Tag[T]


It'd be ideal to take Tag[Emit[Chunk[T]]] evidences directly otherwise the tag macro has to dynamically compose tags, which requires allocation.

fwbrasil · 2024-12-10T00:21:02Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/package.scala

+        subscriber: Subscriber[? >: T]
+    )(
+        using
+        Boundary[Ctx, IO],


We have some inference issues when using Boundary directly. We've been using a workaround with a user-facing inline method calling another private one that then requires the Boundary evidence. See Async for examples.

fwbrasil · 2024-12-10T00:22:16Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamPublisher.scala

+        ): Unit < (Async & Ctx) =
+            Loop(()) { _ =>
+                channel.closed.map {
+                    if _ then


can you use case true => .. case false => ... instead?

fwbrasil · 2024-12-10T00:26:54Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscriber.scala

+
+    override def onSubscribe(subscription: Subscription): Unit =
+        throwIfNull(subscription)
+        var sideEffect: () => Unit = () => ()


would it be possible to avoid the mutable value by using state.get and state.set directly instead of state.update?

yes its can be done, but It'll need to be converted into loop method for correctness

Makes sense. If a Loop is too cumbersome, another option is introducing an update that allows retuning a "tail" computation

fwbrasil · 2024-12-10T00:36:01Z

...reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscription.scala

+        safepoint: Safepoint
+    ): Fiber[Nothing, StreamFinishState] < (IO & Ctx) =
+        boundary { (trace, context) =>
+            val fiber = Fiber.fromTask(IOTask(Poll.run(stream.emit)(poll).map(_._2), safepoint.copyTrace(trace), context))


IOTask is an internal detail of Fiber/Async. Ideally, we should rely on the existing APIs instead of using it directly

this is just a copy of Async._run with boundary that does not care about Abort[E] (E =:= Nothing in this case). Maybe after #906 is merged I can use _run instead

fwbrasil · 2024-12-10T00:41:57Z

build.sbt

+            libraryDependencies ++= Seq(
+                "org.reactivestreams" % "reactive-streams" % "1.0.4",
+                "org.reactivestreams" % "reactive-streams-tck" % "1.0.4" % Test,
+                "org.scalatestplus" %% "testng-7-5" % "3.2.17.0" % Test


Is this required to reuse the tck tests? Would it be possible to use scalatest like the other modules?

We'll need to reimplement those tests in scalatest though.

HollandDM · 2024-12-11T01:45:38Z

@fwbrasil updated, please recheck

HollandDM · 2024-12-11T01:46:52Z

...reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscription.scala

+        ): StreamSubscription[V, Ctx] =
+            _subscribe(stream, subscriber)(subscribeCallback)
+
+        private[kyo] inline def _subscribe[V, Ctx](


Same trick with Async._run, now we don't have to pass context params explicitly

HollandDM · 2024-12-11T01:47:25Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/package.scala

+        Tag[Poll[Chunk[T]]]
+    ): Publisher[T] < (Resource & IO & Ctx) = StreamPublisher[T, Ctx](stream)
+
+    object StreamReactiveStreamsExtensions:


Extension methods added

HollandDM · 2024-12-11T01:48:19Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscriber.scala

+    abstract private[reactivestreams] class SubscriberDone
+    private[reactivestreams] case object SubscriberDone extends SubscriberDone
+
+    enum EmitStrategy derives CanEqual:


introduce EmitStrategy so that users can choose the behavior they want

Nice! We could eventually introduce more advanced strategies including timeouts for example

HollandDM · 2024-12-11T01:49:56Z

...reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscription.scala

+        Frame,
+        Boundary[Ctx, IO & Abort[Nothing]]
+    ): Fiber[Nothing, StreamFinishState] < (IO & Ctx) =
+        Async


replace boundary usage with Async._run

fwbrasil · 2024-12-11T10:12:24Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamSubscriber.scala

+    end onComplete
+
+    private[interop] def await(using Frame): Boolean < Async =
+        @tailrec def handleAwait: Boolean < Async =


Suggested change

@tailrec def handleAwait: Boolean < Async =

@tailrec def handleAwait(): Boolean < Async =

fwbrasil · 2024-12-11T10:14:02Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamPublisher.scala

+        capacity: Int = Int.MaxValue
+    )(
+        using
+        Boundary[Ctx, IO & Abort[Nothing]],


you probably don't need Abort[Nothing] anymore since it's included in IO

fwbrasil · 2024-12-11T10:17:36Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/reactive-streams/StreamPublisher.scala

+                            case Result.Success(_) => Loop.continue(())
+                            case _                 => Loop.done
+                        }
+                }


simplification suggestion:

Loop(()) { _ => Abort.recover[Closed](_ => Loop.done) { for subscriber <- channel.take subscription <- IO.Unsafe(new StreamSubscription[V, Ctx](stream, subscriber)) fiber <- subscription.subscribe.andThen(subscription.consume) _ <- supervisor.onComplete(_ => discard(fiber.interrupt(interruptPanic))) yield Loop.continue(()) } }

fwbrasil · 2024-12-11T10:19:48Z

@HollandDM could you have a look at #918? It seems it could be useful here

fwbrasil · 2024-12-11T10:22:08Z

...eactive-streams/shared/src/test/scala/kyo/interop/reactive-streams/StreamPublisherTest.scala

+import org.reactivestreams.tck.TestEnvironment
+import org.scalatestplus.testng.*
+
+final class StreamPublisherTest extends PublisherVerification[Int](new TestEnvironment(1000L)), TestNGSuiteLike:


I took at the tck tests and it seems they check mostly for correctness without concurrency. If that's the case, I think we'll need to introduce concurrency tests since the code relies heavily on shared state. There are a few examples of concurrency tests like in ChannelTest and MeterTest.

HollandDM · 2024-12-13T10:17:13Z

I moved the implement from extending reactive-streams to extending flow. I think it's more suitable for Kyo

HollandDM · 2024-12-13T10:22:20Z

kyo-reactive-streams/shared/src/test/scala/kyo/interop/flow/SreamSubscriberTest.scala

+    override def cancel(): Unit = isCanceled.set(true)
+end JavaSubscription
+
+final class SreamSubscriberTest extends Test:


I decided to test StreamSubscriber part only, as it's also the only part that use shared atomic state

HollandDM · 2024-12-13T10:24:49Z

kyo-reactive-streams/shared/src/test/scala/kyo/interop/flow/PublisherToSubscriberTest.scala

+        end for
+    }
+
+    "single publisher & multiple subscribers" - {


Added some test between publisher and subscribers

HollandDM · 2024-12-13T10:25:57Z

kyo-reactive-streams/shared/src/test/scala/kyo/interop/flow/SreamSubscriberTest.scala

+import kyo.*
+import kyo.Maybe.*
+
+final class JavaSubscription(subscriber: Subscriber[? >: Int], batchSize: Int, counter: java.util.concurrent.atomic.AtomicInteger)


We need to test a generic Publisher, so I create a generic Java copies

HollandDM · 2024-12-13T15:38:26Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/StreamSubscription.scala

+
+    private[interop] inline def subscribe(using Frame): Unit < IO = IO(subscriber.onSubscribe(this))
+
+    private[interop] def poll(using Tag[Poll[Chunk[V]]], Frame): StreamFinishState < (Async & Poll[Chunk[V]]) =


I tried to convert this into channel.stream(), but this provides more controls over the computation

HollandDM · 2024-12-16T03:06:27Z

@fwbrasil Please recheck if more tests are needed

fwbrasil · 2024-12-16T22:49:05Z

kyo-reactive-streams/shared/src/main/scala/kyo/interop/flow/StreamPublisher.scala

+        Tag[Emit[Chunk[V]]],
+        Tag[Poll[Chunk[V]]]
+    ): StreamPublisher[V, Ctx] < (Resource & IO & Ctx) =
+        inline def interruptPanic = Result.Panic(Fiber.Interrupted(scala.compiletime.summonInline[Frame]))


You won't need the explicit summon after: #929

fwbrasil

This is a sizable contribution, congrats! LGTM but I think you'll need to resolve conflicts with #929

HollandDM · 2024-12-17T03:19:49Z

@fwbrasil thanks, this PR should be good now

HollandDM force-pushed the reactive-stream-interop branch from 2ac4f2e to d3933f9 Compare December 4, 2024 11:47

HollandDM marked this pull request as ready for review December 4, 2024 11:47

HollandDM marked this pull request as draft December 4, 2024 11:48

HollandDM force-pushed the reactive-stream-interop branch 3 times, most recently from d8a986e to 09a9680 Compare December 4, 2024 11:57

HollandDM marked this pull request as ready for review December 4, 2024 11:57

HollandDM mentioned this pull request Dec 4, 2024

feat: STTP stream #880

Draft

HollandDM commented Dec 4, 2024

View reviewed changes

HollandDM force-pushed the reactive-stream-interop branch from 09a9680 to fe3799a Compare December 6, 2024 11:20

HollandDM commented Dec 6, 2024

View reviewed changes

calvinlfer reviewed Dec 6, 2024

View reviewed changes

fwbrasil reviewed Dec 9, 2024

View reviewed changes

fwbrasil reviewed Dec 10, 2024

View reviewed changes

HollandDM force-pushed the reactive-stream-interop branch 2 times, most recently from 8933493 to fb04b3a Compare December 11, 2024 01:45

HollandDM requested a review from fwbrasil December 11, 2024 01:45

HollandDM commented Dec 11, 2024

View reviewed changes

fwbrasil reviewed Dec 11, 2024

View reviewed changes

HollandDM force-pushed the reactive-stream-interop branch from fb04b3a to 2a28269 Compare December 13, 2024 09:54

HollandDM requested a review from fwbrasil December 13, 2024 10:13

HollandDM commented Dec 13, 2024

View reviewed changes

HollandDM force-pushed the reactive-stream-interop branch from 2a28269 to 781353b Compare December 13, 2024 15:36

HollandDM commented Dec 13, 2024

View reviewed changes

fwbrasil reviewed Dec 16, 2024

View reviewed changes

fwbrasil approved these changes Dec 16, 2024

View reviewed changes

HollandDM added 4 commits December 17, 2024 09:58

reactivestream/flow interop

b2b1b6f

fix: using poll

ca4a01a

[kyo-reactive-stream]: fix feedbacks

91ed5bd

adjust KyoException

79cdc47

HollandDM force-pushed the reactive-stream-interop branch from 781353b to 79cdc47 Compare December 17, 2024 03:18

HollandDM requested a review from fwbrasil December 17, 2024 03:34

fwbrasil merged commit 702fbdc into getkyo:main Dec 17, 2024
3 checks passed

fwbrasil mentioned this pull request Dec 17, 2024

Reactive streams TCK tests are too slow #938

Open


		private[reactivestreams] inline def subscribe: Unit < IO = IO(subscriber.onSubscribe(this))

		private[reactivestreams] def poll: StreamFinishState < (Async & Poll[Chunk[V]]) =

	@tailrec def handleAwait: Boolean < Async =
	@tailrec def handleAwait(): Boolean < Async =


		private[interop] inline def subscribe(using Frame): Unit < IO = IO(subscriber.onSubscribe(this))

		private[interop] def poll(using Tag[Poll[Chunk[V]]], Frame): StreamFinishState < (Async & Poll[Chunk[V]]) =

Reactive-stream/Flow Interop #896

Reactive-stream/Flow Interop #896

Conversation

HollandDM commented Dec 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HollandDM Dec 10, 2024 • 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

HollandDM Dec 11, 2024 • edited Loading

Choose a reason for hiding this comment

fwbrasil Dec 11, 2024 • 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

HollandDM Dec 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HollandDM Dec 10, 2024 • edited Loading

Choose a reason for hiding this comment

HollandDM commented Dec 11, 2024

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

fwbrasil commented Dec 11, 2024

fwbrasil Dec 11, 2024 • edited Loading

Choose a reason for hiding this comment

HollandDM commented Dec 13, 2024

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

HollandDM commented Dec 16, 2024

Choose a reason for hiding this comment

fwbrasil left a comment

Choose a reason for hiding this comment

HollandDM commented Dec 17, 2024

HollandDM Dec 10, 2024 •

edited

Loading

HollandDM Dec 11, 2024 •

edited

Loading

fwbrasil Dec 11, 2024 •

edited

Loading

HollandDM Dec 10, 2024 •

edited

Loading

HollandDM Dec 10, 2024 •

edited

Loading

fwbrasil Dec 11, 2024 •

edited

Loading