Add interop.flow.pipeToProcessor
#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.
BalmungSan
changed the title
interop.flow.pipeToProcessorinterop.flow.pipeToProcessor
| * like requesting elements from a database. | ||
| * A high number will also lead to more elements in memory. | ||
| */ | ||
| def pipeToProcessor[F[_], I, O]( |
There was a problem hiding this comment.
We can add these as syntax on Pipe too :)
fs2/core/shared/src/main/scala/fs2/Stream.scala
Line 5494 in 0377f21
| ): 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?
Convenient helper.
Roadmap:
unsafeversion.