Add ZStream#zipWithLatest #1482

regiskuckaertz · 2019-08-25T17:16:18Z

Resolves #1204
Resolves #1492

Happy to close this if #1270 is updated (cc @contrun )

I have also added a few combinators I couldn't find how to express easily, but if I missed something I'll get rid of them.

…test

iravid

Needs a quick rebase after the rename of InputStream to Pull. Also left two minor comments.

streams/shared/src/main/scala/zio/stream/ZStream.scala

streams-tests/jvm/src/test/scala/zio/stream/StreamSpec.scala

…test

streams/shared/src/main/scala/zio/stream/ZStream.scala

project/BuildHelper.scala

iravid · 2019-08-26T17:57:15Z

streams/shared/src/main/scala/zio/stream/ZStream.scala

@@ -1530,6 +1537,24 @@ class ZStream[-R, +E, +A](val process: ZManaged[R, E, Pull[R, E, A]]) extends Se
   */
  final def runDrain: ZIO[R, E, Unit] = run(Sink.drain)

+  final def schedule[R1 <: R, E1 >: E, B](schedule: ZSchedule[R1, A, B]): ZStream[R1 with Clock, E1, B] =


This is really nice. Can it also replace spaced?

Yeah I'll try to 👍

There's some weird behaviour when I use a schedule, where the second stream in my test makes progress much faster than the first one in the init phase. I will investigate.

Found it. The schedule will now contribute to deciding the end of the stream. So, if the schedule ends after 2 repeats, only 3 elements will be produced. It seems more intuitive to me (see the definition of fixed), but I'm keen to hear your opinion. If ok I'll move on to look at spaced.

I actually found the previous behavior, in which decision.cont == false means that the schedule is restarted, to be more intuitive. This is also more in line with aggregateWithin, which restarts the schedule when it stops.

Could you explain what's the advantage?

By the way, we can attain this behavior with the previous formulation if you emit the schedule's output into the stream. We can then use takeWhile or collectWhile to end the stream when the schedule ends.

Hehe, I had a realization myself. There are two ways to apply a schedule to a stream with regards to the schedule's cont.

Given a stream A B C and a schedule (Schedule.exponential(2.millis) && Schedule.recur(2) && Schedule.const("Done")).map(_._2), these are the interpretations:

As long as the schedule wants to continue, repeat the current element after a delay:
A |-2ms-| A |-4ms-| A |-8ms-| "Done" B |-2ms-| B |-4ms-| B |-8ms-| "Done" ...

As long as the schedule wants to continue, emit the stream's next element after a delay:
A |-2ms-| B |-4ms-| C |-8ms-| "Done"

The first interpretation is the existing spaced. I guess that this implementation should be the second interpretation.

dang, I was quite happy with the terseness of scheduleWith 😅 there may be some confusion as to which does what, are you ok with naming the one that acts on each elements scheduleElementsWith and the one that acts on the whole stream scheduleWith (I also think it would fix the confusion between ZStream.spaced and ZSchedule.spaced)?

Yeah, those names are good! We can bring ZStream#repeat inline with that too:

The signature can be changed to:

final def repeat[R1 <: R, B](schedule: ZSchedule[R1, Any, B]): ZStream[R1 with Clock, E, Either[B, A]]

And a repeatWith variant can be added.

Can be done in a follow-up of course :-)

Let's do in a follow-up, I'd like to see where the ticket on combinators is going.

Ending with the schedule is more "foundational", since you can turn a terminating schedule into a forever schedule with Schedule#forever. However it is less convenient. Usually this argues for two methods: one to do the more essential thing, and the other, more visible and obvious one, to do the Schedule#forever thing.

quelgar · 2019-08-29T07:30:53Z

streams/shared/src/main/scala/zio/stream/ZStream.scala

+      for {
+        is    <- self.mergeEither(that).process
+        state <- Ref.make[(Option[A], Option[B])]((None, None)).toManaged_
+        pull: Pull[R1, E1, C] = {


Nit: you could just put the pull expression directly in the yield.

It's kind of a pattern in the ZStream class, I think it's a little easier to read.

regiskuckaertz · 2019-08-29T08:56:41Z

@iravid I think this is ready for a final review, other schedule-based combinators can be documented in the ticket? I'm happy to pick them up right after... but I wonder if sth more useful would be to convert tests to the new framework?

iravid · 2019-08-29T13:18:38Z

@regiskuckaertz Help with the tests conversion would be much much appreciated 🙏🏻 but really, pick up whatever you fancy the most :-)

iravid · 2019-08-31T06:10:57Z

@regiskuckaertz could you amend a commit to re-trigger the CI?

iravid · 2019-08-31T06:24:41Z

streams/shared/src/main/scala/zio/stream/ZStream.scala

+                        for {
+                          decision <- schedule.update(a, sched)
+                          _        <- clock.sleep(decision.delay)
+                          _        <- state.set((!decision.cont, decision.state, Some(decision.finish)))


Hmm didn't we say that the schedule should not stop the stream?

Ah I see that my example was unclear. In this formulation, I meant that the schedule should restart between the elements. So for a stream "A B C A B C", this should be result:

A |-2ms-| B |-4ms-| C |-8ms-| "Done" A |-2ms-| B |-4ms-| C |-8ms-| "Done"

Nope, this is implement for point 2 above:

As long as the schedule wants to continue, emit the stream's next element after a delay

And the short circuiting can be attained by collectWhile { case Right(v) => v }, which would stop on the first Left.

(sorry for dragging this!)

Actually, let's merge this and address in a follow-up.

That is the semantics of scheduleElements, isn't it? 🤔 I could add a test to prove that.

Ah not quite. scheduleElements repeats every element and restarts

regiskuckaertz · 2019-08-31T06:44:21Z

Thanks 🙏

Regis Kuckaertz added 9 commits August 25, 2019 07:25

Change console init for stream project

629bd7c

Breaking the 4th wall :)

f7f6db8

Add a test

39d942f

Add convenience methods iterate and delay

07b647a

Reorder methods alphabetically

681d5dd

Add zipWithLatest

7e9ca68

Merge branch 'master' of https://github.com/zio/zio into rk-zipwithla…

327d593

…test

Add scaladoc

ad011d9

Simpler formulation

98e725c

regiskuckaertz requested a review from iravid August 25, 2019 17:17

fmt

3c1abf0

iravid reviewed Aug 25, 2019

View reviewed changes

streams-tests/jvm/src/test/scala/zio/stream/StreamSpec.scala Show resolved Hide resolved

Regis Kuckaertz added 4 commits August 25, 2019 19:47

Scaladoc

4f09aa2

Use latch

4a8e06e

Merge branch 'master' of https://github.com/zio/zio into rk-zipwithla…

b1ff063

…test

Use Pull and new apply

9847b62

jdegoes reviewed Aug 25, 2019

View reviewed changes

streams/shared/src/main/scala/zio/stream/ZStream.scala Show resolved Hide resolved

iravid reviewed Aug 26, 2019

View reviewed changes

project/BuildHelper.scala Show resolved Hide resolved

Define schedule and rewrite delay using it

e88eccb

iravid reviewed Aug 26, 2019

View reviewed changes

iravid mentioned this pull request Aug 26, 2019

Add combinator to tick elements of a ZStream at regular interval #1492

Closed

Regis Kuckaertz added 6 commits August 27, 2019 21:35

Init schedule for each element

f77c5ee

Better

9a0bd1c

fmt

791a5c1

Use schedule per element

f9c8db2

fmtttttt

2cc02a8

scheduleWith

9637639

quelgar reviewed Aug 29, 2019

View reviewed changes

schedule* and scheduleElements*

71ca0ac

scaladoc

2db075d

iravid reviewed Aug 31, 2019

View reviewed changes

iravid approved these changes Aug 31, 2019

View reviewed changes

iravid merged commit 88eb7e4 into zio:master Aug 31, 2019

regiskuckaertz deleted the rk-zipwithlatest branch August 31, 2019 06:42

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ZStream#zipWithLatest #1482

Add ZStream#zipWithLatest #1482

regiskuckaertz commented Aug 25, 2019 •

edited by iravid

iravid left a comment

iravid Aug 26, 2019

regiskuckaertz Aug 26, 2019

regiskuckaertz Aug 27, 2019

regiskuckaertz Aug 27, 2019

iravid Aug 28, 2019

iravid Aug 29, 2019

regiskuckaertz Aug 29, 2019 •

edited

iravid Aug 29, 2019

regiskuckaertz Aug 30, 2019

jdegoes Aug 31, 2019

quelgar Aug 29, 2019

regiskuckaertz Aug 29, 2019

regiskuckaertz commented Aug 29, 2019

iravid commented Aug 29, 2019

iravid commented Aug 31, 2019

iravid Aug 31, 2019

iravid Aug 31, 2019

regiskuckaertz Aug 31, 2019

iravid Aug 31, 2019

iravid Aug 31, 2019

iravid Aug 31, 2019

regiskuckaertz Aug 31, 2019

iravid Aug 31, 2019

regiskuckaertz commented Aug 31, 2019

Add ZStream#zipWithLatest #1482

Add ZStream#zipWithLatest #1482

Conversation

regiskuckaertz commented Aug 25, 2019 • edited by iravid

iravid left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

regiskuckaertz Aug 29, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

regiskuckaertz commented Aug 29, 2019

iravid commented Aug 29, 2019

iravid commented Aug 31, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

regiskuckaertz commented Aug 31, 2019

regiskuckaertz commented Aug 25, 2019 •

edited by iravid

regiskuckaertz Aug 29, 2019 •

edited