Add interop.flow.pipeToProcessor & interop.flow.processorToPipe
#3449
Conversation
| ): Resource[F, StreamProcessor[F, I, O]] = | ||
| for { | ||
| streamSubscriber <- Resource.eval(StreamSubscriber[F, I](chunkSize)) | ||
| inputStream = streamSubscriber.stream(subscribe = F.unit) |
There was a problem hiding this comment.
We don't have access to the upstream Publisher yet, so we can't do the subscribe here. So we do nothing and wait for it.
There was a problem hiding this comment.
So we do nothing and wait for it.
I'm not entirely sure I understand the waiting part. Can/should we use a Deferred here?
There was a problem hiding this comment.
No sorry, my point is that subscribe should be something like IO(publisher.subscribe(streamSubscriber)) as you can see in the implementation of syntax.fromPublisher.
Here, however, we don't have a Publisher yet, so we don't control when we are published.
Basically, this is very similar to the fromPublisher overload that accepts a Subscriber => F[Unit], just that rather than receiving the lambda, we are just returning a raw Processor / Subscriber.
There was a problem hiding this comment.
Thanks, I understand now.
I think what confused me was that this is the first time we are effectively exposing StreamSubscriber. Our existing {to,from}Publisher APIs have kept it hidden.
So I do feel weird about exposing it now. But I guess nothing here is unreasonable 🤔
interop.flow.pipeToProcessorinterop.flow.pipeToProcessor
| ): Resource[F, StreamProcessor[F, I, O]] = | ||
| for { | ||
| streamSubscriber <- Resource.eval(StreamSubscriber[F, I](chunkSize)) | ||
| inputStream = streamSubscriber.stream(subscribe = F.unit) |
There was a problem hiding this comment.
So we do nothing and wait for it.
I'm not entirely sure I understand the waiting part. Can/should we use a Deferred here?
24a844d to
d59c5b8
Compare
| final class ProcessorPipeSpec extends Fs2Suite { | ||
| test("should process upstream input and propagate results to downstream") { | ||
| forAllF(Arbitrary.arbitrary[Seq[Int]], Gen.posNum[Int]) { (ints, bufferSize) => | ||
| // Since creating a Flow.Processor is very complex, | ||
| // we will reuse our Pipe => Processor logic. | ||
| val processor = unsafePipeToProcessor[Int, Int]( | ||
| pipe = stream => stream.map(_ * 1), | ||
| chunkSize = bufferSize | ||
| ) |
There was a problem hiding this comment.
Not sure if there is a better way to test this.
Also, not sure if there is something else we may test.
interop.flow.pipeToProcessorinterop.flow.pipeToProcessor & interop.flow.processorToPipe
f2a1c55 to
a67b91f
Compare
|
CI passed this time, but one of the tests seems flaky. Possible race condition? https://github.com/BalmungSan/fs2/actions/runs/12014747489/job/33491269929#step:11:1615 |
|
@armanbilge I have been re-running the test for a while now, and it always succeeds in a couple of seconds. |
I see the exact same CI failure in your latest merge commit 😕 it must be something legit. I will try to reproduce ... https://github.com/typelevel/fs2/actions/runs/12967866568/job/36170223920?pr=3449#step:11:1615 |
7d2952b to
64194f0
Compare
Convenient helpers.
Roadmap:
unsafeversion.