Move j.u.c.Flow interop methods onto Stream
#3346
Conversation
| subscription.run | ||
| ): Stream[F, Nothing] = | ||
| Stream.eval(apply(stream, subscriber)).flatMap { subscription => | ||
| Stream.eval(F.delay(subscriber.onSubscribe(subscription))) >> subscription.run |
There was a problem hiding this comment.
This should be exec and ++, shouldn't it?
There was a problem hiding this comment.
If it's exec, then subscription.run will never evaluate.
The two options are:
Stream.eval(...) >> subscription.run
Stream.exec(...) ++ subscription.runThe former felt more appropriate to me, it establishes the monadic dependence. Not sure what the subtle differences are.
| /** Provides syntax for `IO` streams. */ | ||
| implicit final class IOOps[A](private val self: Stream[IO, A]) extends AnyVal { | ||
|
|
||
| /** Creates a [[Publisher]] from this [[Stream]]. | ||
| * | ||
| * The stream is only ran when elements are requested. | ||
| * | ||
| * @note This [[Publisher]] can be reused for multiple [[Subscribers]], | ||
| * each [[Subscription]] will re-run the [[Stream]] from the beginning. | ||
| * | ||
| * @see [[toPublisher]] for a safe version that returns a [[Stream]]. | ||
| */ | ||
| def unsafeToPublisher()(implicit | ||
| runtime: IORuntime | ||
| ): Publisher[A] = | ||
| interop.flow.unsafeToPublisher(self) | ||
| } |
There was a problem hiding this comment.
This is the one I'm most interested in. It is the Stream analog of io.unsafeToFuture().
The motivation is to facilitate interop with other libraries e.g. raquo/Airstream#114:
EventStream.fromPublisher(fs2Stream.unsafeToPublisher())
|
A couple of random thoughts against this:
A couple of random thoughts in favor of this:
|
|
I think those are really good points, which is why I asked Luis to post them here. I think a reasonable compromise could be to keep |
So that it Just Works™️ without the annoying
import fs2.interop.flow.syntax._, which is now deprecated. cc @BalmungSan