Adding ZSink#zipPar
#1344
Adding ZSink#zipPar
#1344
Conversation
|
@iravid could you please take a look? |
| ZIO.succeed(Right(Step.state(st))) | ||
| } else { | ||
| self | ||
| .extract(Step.state(st)) |
There was a problem hiding this comment.
Curious, why not defer the extraction to extract?
| } | ||
|
|
||
| override def initial: ZIO[R1, E1, Step[State, Nothing]] = | ||
| self.initial.flatMap(s1 => { |
There was a problem hiding this comment.
2 minor notes:
- these types of blocks usually read nicer as
self.initial.flatMap { s1 => - whenever you have multiple consecutive
flatMaps, those usually read nicer as aforcomprehension
There was a problem hiding this comment.
I do agree about using for and initially tried to write it like that. But I faced no method withFilter error and then continued with flatMaps
| r2 match { | ||
| case Left((c, rem2)) => { | ||
| val minLeftover = | ||
| if (rem1.isEmpty && rem2.isEmpty) Chunk.empty else (rem1.toList ++ rem2.toList).minBy(_.length) |
There was a problem hiding this comment.
Converting the remainders to a list could be potentially costly, so let's do this the old fashioned way by comparing the lengths with an if
There was a problem hiding this comment.
Maybe it's not obvious but rem1 and rem2 here are Option[Chunk], not Chunk. Option#toList it's relatively cheap and this code runs only once.
The logic is following: if one sink terminated strictly before the other I drop remainder of the former anyway. I compare remainders' length only if sinks finished simultaneously.
But if you still think it's costly - I'll change it.
| @@ -167,6 +167,9 @@ class SinkSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) | |||
| zipLeft (<*) | |||
| happy path $zipLeftHappyPath | |||
|
|
|||
| zipPar | |||
| zipPar somehow works $zipParSanity | |||
There was a problem hiding this comment.
Here are some additional test cases I suggest we add:
- left sink finishes earlier
- right sink finishes earlier
- both sinks finish at the same time
- left sink errors during step
- left sink errors during extract
- right sink errors during step
- right sink errors during extract
There was a problem hiding this comment.
Yeah. Must cover cases of non-terminating sinks as well. And also it would be nice to explicitly check the logic of choosing the shorter leftover when sinks terminated at the same time.
Haven't found test sinks with non-trivial leftovers to use here. Will invent one.
=====
Actually there is no need to manually test sink1.zipPar(sink2) and then sink2.zipPar(sink1). Swapping the input swaps the output. Just need to verify this law.
|
@iravid thank you for kind words! |
|
@iravid I think I'm done with it. |
| @@ -780,6 +818,116 @@ class SinkSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) | |||
| unsafeRun(sinkIteration(sink, 1).map(_ must_=== "1Hello")) | |||
| } | |||
|
|
|||
| private object ZipParLaws { | |||
There was a problem hiding this comment.
This is excellent, comprehensive work! Big like :-)
#1310
Added a combinator for aggregating a stream into two sinks at the same time.
Please review my approach. I'll add more tests and squash commits afterwards.