add ZStream#{groupBy, split, fanOut} #1387
Conversation
iravid
changed the title
There was a problem hiding this comment.
Amazing work @mschuwalow. I left a first round of comments below. Need to spend some more time with groupBy.
Few notes:
- not entirely sold on
flatMapParUnboundedandflatMapParSema- these need a bit more justification - we'll wait with getting this in until we address the encoding, so you'll have a chance to kick the tires on a simpler representation of stream
- the infrastructure in
toQueuesBalanced0is really impressive. I'd be very interested in seeing the consistent hashing implementation. This is very close to a good general purpose PubSub implementation; the only thing it is currently missing is dynamic subscribe/unsubscribe.
Co-Authored-By: Itamar Ravid <iravid@users.noreply.github.com>
…uwalow/scalaz-zio into feature/zio#1378-add-ZStream#fanOut
|
i think this is ready for another review |
There was a problem hiding this comment.
Let's rebase this and give this another look afterwards. Sorry for taking so long @mschuwalow !
| @@ -791,7 +791,7 @@ object ZManaged { | |||
| * specified text message. This method can be used for terminating a fiber | |||
| * because a defect has been detected in the code. | |||
| */ | |||
| final def dieMessage(message: String): ZManaged[Any, Throwable, Nothing] = die(new RuntimeException(message)) | |||
| final def dieMessage(message: String): ZManaged[Any, Nothing, Nothing] = die(new RuntimeException(message)) | |||
| queues <- queues.get | ||
| _ <- ZIO.foreach_(queues) { q => | ||
| q.offer(a).catchAllCause { | ||
| case c if (c.interrupted) => ZIO.unit |
There was a problem hiding this comment.
Partial function used where total one expected...
There was a problem hiding this comment.
Ugh that was close. Shameful 😨
I brainfarted and thought this was defined as a partialfunction on causes, halting with any that don't match
|
@iravid please take a look at the new signature of groupby. |
There was a problem hiding this comment.
A few minor changes and this is good to go. Amazing work @mschuwalow!
Great. I think you can take another look when you have the time :) |
There was a problem hiding this comment.
Thank you @mschuwalow! Merging this now; could you follow up with the scaladoc adjustments? Thanks!
| @@ -542,6 +569,75 @@ trait ZStream[-R, +E, +A] extends Serializable { self => | |||
| final def concat[R1 <: R, E1 >: E, A1 >: A](other: => ZStream[R1, E1, A1]): ZStream[R1, E1, A1] = | |||
| ZStream(UIO.succeed(self), UIO(other)).flatMap(ZStream.unwrap) | |||
|
|
|||
| /** | |||
| * More powerful version of `ZStream#toQueues`. Allows to provide a function that determines what | |||
There was a problem hiding this comment.
Sure. Will try to do it tomorrow :)
| } | ||
|
|
||
| /** | ||
| * More powerful version of `ZStream#toQueuesBalanced`. This returns a function that will produce |
Resolves #1378
Resolves #1102
@iravid Would be great if you could take a look at this.
Based on the toQueuesManaged (naming things is hard, I'm very open to suggestions :D) one can implent quite a bit more powerful stuff. In fact I had a version with a partioner using a consitent hashing scheme that allowed adhoc up and down scaling. I don't think this is paticularly useful as is, but might be interesting to explore later