Add Pipeline conversions and rewrite internals with channels #373
Conversation
| () | ||
| } | ||
| } | ||
| )(implicit trace: Trace): ZIO[R, Nothing, Publisher[O]] = ZIO.runtime[R].map { runtime => subscriber => |
There was a problem hiding this comment.
This guy cannot be implemented on top of channelToPublisher due to the missing R <: Scope constraint
There was a problem hiding this comment.
How do you feel about just adding the constraint? It would be more correct if the fork was scoped IMO.
| unsafe { implicit unsafe => | ||
| val subscription = new SubscriptionProducer[I](subscriber) | ||
|
|
||
| def reader: ZChannel[Any, ZNothing, Chunk[I], Any, Nothing, Chunk[I], Unit] = ZChannel.readWithCause( |
There was a problem hiding this comment.
this needs a custom implementation because we need to prevent defects from getting send to it
| )(implicit trace: Trace): ZPipeline[Any, Throwable, I, O] = ZPipeline.unwrapScoped { | ||
| val subscription = new SubscriptionProducer[I](processor)(unsafe) | ||
| val subscriber = new SubscriberConsumer[O](bufferSize)(unsafe) | ||
| val passthrough = new PassthroughAsyncInput(subscription, subscriber) |
There was a problem hiding this comment.
needs a custom implementation so defects are not passed
|
Bumping this, as it would really help to implement WebSocket support for |
|
This PR also makes cancellations of streams work better, previously cancellations only seemed to complete the stream if read after closing the stream. |
There was a problem hiding this comment.
Thank you for this awesome contribution @mschuwalow, and sorry it took so loong before I noticed.
Just a few tiny comments.
| () | ||
| } | ||
| } | ||
| )(implicit trace: Trace): ZIO[R, Nothing, Publisher[O]] = ZIO.runtime[R].map { runtime => subscriber => |
There was a problem hiding this comment.
How do you feel about just adding the constraint? It would be more correct if the fork was scoped IMO.
| if (!isSubscribedOrCanceled.compareAndSet(false, true)) { | ||
| s.cancel() | ||
| } else { | ||
| subscription.unsafe.done(ZIO.succeedNow(s)) |
There was a problem hiding this comment.
succeedNow is deprecated. Let's switch to succeed.
Reason to go this way is that for pipeline to work correctly, we need to ensure that the Publisher and the Subscriber run concurrently on different fibers. Imo this is easiest / cleanest by relying on the AsynInput abstractions baked into the channelexecutor.
resolves #364