Pipe #1166
Pipe #1166
Conversation
| * @return | ||
| * A new stream of transformed element type `A` | ||
| */ | ||
| def into[A, S2](pipe: Pipe[V, A, S2])(using Tag[V], Frame): Stream[A, S & S2] = |
There was a problem hiding this comment.
how about map instead of into? Pipe is similar to a map function
There was a problem hiding this comment.
It's similar but different in important ways. map makes it seem like each element will be transformed, which is not necessarily the case. The input and output streams can be entirely different lengths, for instance.
| * @return | ||
| * New pipe that performs both transformations in sequence | ||
| */ | ||
| def join[C, S1](pipe: Pipe[B, C, S1])(using Tag[B], Frame): Pipe[A, C, S & S1] = |
There was a problem hiding this comment.
Would this make sense as andThen? I think we have some other APIs with that naming.
There was a problem hiding this comment.
I prefer not to re-use extension method names as this
8000
can create annoying conflicts (Pipe[A, B, S] can be lifted to Pipe[A, B, S] < Any, the andThen extension on < could be accessed).
|
Looks like the failed build is unrelated to this PR |
|
@johnhungerford it's a draft, or ready to review + merge? (it looks good to me) |
|
Ready for review as far as I'm concerned |
| def contramap[A1, S1](f: A1 => A < S1)( | ||
| using | ||
| t1: Tag[Poll[Chunk[A]]], | ||
| t2: Tag[Poll[Chunk[A1]]], | ||
| disc: Discriminator, | ||
| fr: Frame | ||
| ): Pipe[A1, B, S & S1] = | ||
| Pipe: | ||
| ArrowEffect.handleLoop(t1, pollEmit)( | ||
| [C] => | ||
| (unit, cont) => | ||
| Poll.andMap[Chunk[A1]]: | ||
| case Absent => Loo 8000 p.continue(cont(Absent)) | ||
| case Present(chunk) => | ||
| Kyo.foreach(chunk)(f).map: chunk2 => | ||
| Loop.continue(cont(Present(chunk2))) | ||
| ) |
There was a problem hiding this comment.
I guess it's going to be contramapKyo
Completes #1124
Problem
It would be good to have a Kyo version of ZIO's
ZPipelineSolution
As discussed in #1124 , added
Pipe[A, B, S], a wrapper aroundUnit < (Poll[Chunk[A]] & Emit[Chunk[B]] & S).Combinators:
Constructors:
You can now process a stream via a sequence of pipes terminating with a sink:
val processed = stream.into(pipe1).into(pipe2).into(pipe3).into(sink)Notes
Pipe.transformandSink.drainno longer usPoll.runsince this method handledEmitprior toPoll.PipeandSinkdelegate control to their ownPolleffects, allowing early termination.Stream, since currentPipes are all based onStreammethods. Users should expect identical behavior.