8000 Restart schedule on end by regiskuckaertz · Pull Request #1767 · zio/zio · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

Restart schedule on end #1767

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Sep 22, 2019
Merged

Restart schedule on end #1767

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
38e5b9e
Restart schedule on end
Sep 22, 2019
eeea8e7
Merge branch 'master' of https://github.com/zio/zio into rk-stream-sc…
Sep 22, 2019
e3c9ec9
Remove unused state, add docs
Sep 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions streams-tests/jvm/src/test/scala/zio/stream/StreamSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1600,16 +1600,16 @@ class StreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestRu

private def scheduleWith =
unsafeRun(
Stream("A", "B", "C")
.scheduleWith(Schedule.recurs(3) *> Schedule.fromFunction((_) => "!"))(_.toLowerCase, identity)
Stream("A", "B", "C", "A", "B", "C")
.scheduleWith(Schedule.recurs(2) *> Schedule.fromFunction((_) => "Done"))(_.toLowerCase, identity)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm why did it change to 2?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Schedule.recurs(2) will consume three elements, as in the test for scheduleWith

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok let me wrap my head around this for a sec..running your state machine manually ;)

.run(Sink.collectAll[String])
.map(_ must_=== List("a", "b", "c", "!"))
.map(_ must_=== List("a", "b", "c", "Done", "a", "b", "c", "Done"))
)

private def scheduleEither =
unsafeRun(
Stream("A", "B", "C")
.scheduleEither(Schedule.recurs(3) *> Schedule.fromFunction((_) => "!"))
.scheduleEither(Schedule.recurs(2) *> Schedule.fromFunction((_) => "!"))
.run(Sink.collectAll[Either[String, String]])
.map(_ must_=== List(Right("A"), Right("B"), Right("C"), Left("!")))
)
Expand Down
47 changes: 28 additions & 19 deletions streams/shared/src/main/scala/zio/stream/ZStream.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1810,26 +1810,35 @@ class ZStream[-R, +E, +A](val process: ZManaged[R, E, Pull[R, E, A]]) extends Se
for {
as <- self.process
init <- schedule.initial.toManaged_
state <- Ref.make[(Boolean, schedule.State, Option[() => B])]((false, init, None)).toManaged_
state <- Ref.make[(schedule.State, Option[() => B])]((init, None)).toManaged_
pull = state.get.flatMap {
case (done, sched, decision0) =>
if (done) Pull.end
else
for {
a <- as.optional.mapError(Some(_))
c <- a match {
case Some(a) =>
for {
decision <- schedule.update(a, sched)
_ <- clock.sleep(decision.delay)
_ <- state.set((!decision.cont, decision.state, Some(decision.finish)))
} yield if (decision.cont) f(a) else g(decision.finish())

case None =>
state.set((false, sched, None)) *> decision0
.fold[Pull[R1 with Clock, E1, C]](Pull.end)(b => Pull.emit(g(b())))
}
} yield c
case (sched0, finish0) =>
// Before pulling from the stream, we need to check whether the previous
// action ended the schedule, in which case we must emit its final output
finish0 match {
case None =>
for {
a <- as.optional.mapError(Some(_))
c <- a match {
// There's a value emitted by the underlying stream, we emit it
// and check whether the schedule ends; in that case, we record
// its final state, to be emitted during the next pull
case Some(a) =>
for {
decision <- schedule.update(a, sched0)
_ <- clock.sleep(decision.delay)
sched <- if (decision.cont) UIO.succeed(decision.state) else schedule.initial
finish = if (decision.cont) None else Some(decision.finish)
_ <- state.set((sched, finish))
} yield f(a)

// The stream ends when both the underlying stream ends and the final
// schedule value has been emitted
case None => Pull.end
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This means that we don't emit the schedule's final value if the stream ended before the schedule ended. Is that intentional?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. If for instance the stream is completely empty, there never was a value of type a to update the schedule.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I'm good with that 👍🏻

}
} yield c
case Some(b) => state.set((sched0, None)) *> Pull.emit(g(b()))
}
}
} yield pull
}
Expand Down
0