Michael Pilquist mpilquist

TL;DR - Use fs2.time.sleep_[Task](delay) ++ Stream.eval(effect) instead of Stream.eval(effect.schedule(delay)).

FS2 never interrupts evaluation of an effect. This can lead to surprising behavior when using the schedule method on Task. Consider this test driver:

def testInterruption[A](effect: Stream[Task, A]): Stream[Task, A] = {
  val logStart = Stream.eval_(Task.delay(println("Started: " + System.currentTimeMillis)))
  val logFinished = Stream.eval_(Task.delay(println("Finished: " + System.currentTimeMillis)))
  val interruptSoonAfterStart =
 Stream.eval(async.signalOf[Task,Boolean](false)).flatMap { cancellationSignal =&gt;

FS2 0.9.4 Release Announcement

FS2 0.9.4 is now available!

The 0.9.3 release was so much fun, we decided to release 0.9.4 too, this time with no memory leaks introduced by sloppy git merge conflict resolutions (sorry!).

0.9.4 is backwards compatible with earlier releases in the 0.9 series.

See the changelog and the 0.9.3 release announcement for more details.

	/*
	scalaVersion := "2.12.7"

	resolvers += Resolver.sonatypeRepo("snapshots")

	libraryDependencies += "co.fs2" %% "fs2-core" % "1.0.1-SNAPSHOT"
	*/
	import cats._
	import cats.implicits._
	import cats.effect._

	// Range is a product of two ints that implicitly carries a proof that start <= end.
	// We can model this in scala as a case class with a validating apply method.
	// Note: requires relatively recent 2.12 release

	final case class Range private (start: Int, end: Int) {
	def copy(start: Int = this.start, end: Int = this.end): Option[Range] = Range(start, end)
	}
	object Range {
	def apply(start: Int, end: Int): Option[Range] =
	if (start <= end) Some(new Range(start, end)) else None

	import $ivy.`org.typelevel::cats-effect:0.10-22efb61`, cats._, cats.implicits._, cats.effect._, scala.concurrent.duration._

	def putStr(msg: String): IO[Unit] = IO(print(msg))
	def putStrLn(msg: String): IO[Unit] = IO(println(msg))
	def readLn: IO[String] = IO(Console.readLine)

	val now: IO[Long] = IO(System.currentTimeMillis)

	def time[A](ioa: IO[A]): IO[(FiniteDuration, A)] = for {
	start <- now

	import $ivy.`org.typelevel::cats-effect:0.10-22efb61`

	def disablePrettyPrintIO = repl.pprinter.update(repl.pprinter().copy(additionalHandlers = {
	case io: cats.effect.IO[_] => pprint.Tree.Literal("✨✨✨")
	}))

	disablePrettyPrintIO

	def enablePrettyPrintIO = repl.pprinter.update(repl.pprinter().copy(additionalHandlers = PartialFunction.empty))

	object Example {
	trait H[A]

	def id[A]: H[A] => H[A] = in => in

	def apply[A,B](h: H[A], f: H[A] => H[B]): H[B] = f(h)
	def applyIdExplicitType[A](h: H[A]): H[A] = apply(h, id)

	// Fails in 2.11,2.12,2.13-M2, works in Dotty
	// found : Example.H[Nothing] => Example.H[Nothing]

	package fs3

	import cats.Monad

	sealed abstract class Free[F[_], R] {
	def flatMap[R2](f: R => Free[F, R2]): Free[F, R2] = Free.Bind(this, f)
	}
	object Free {
	case class Pure[F[_], R](r: R) extends Free[F, R]
	case class Eval[F[_], R](fr: F[R]) extends Free[F, R]

	// Like interruptWhen / interrupt but allows resumption of the source stream.

	def pauseWhen[F[_]: Async, A](controlSignal: Signal[F, Boolean]): Pipe[F, A, A] =
	pause(controlSignal.discrete)

	def pause[F[_]: Async, A](control: Stream[F, Boolean]): Pipe[F, A, A] = {
	def unpaused(
	controlFuture: ScopedFuture[F, Pull[F, Nothing, (NonEmptyChunk[Boolean], Handle[F, Boolean])]],
	srcFuture: ScopedFuture[F, Pull[F, Nothing, (NonEmptyChunk[A], Handle[F, A])]]
	): Pull[F, A, Nothing] = {

	def switchMap[F[_]: Async, A, B](f: A => Stream[F, B]): Pipe[F, A, B] = {
	def go(
	outer: ScopedFuture[F, Pull[F, Nothing, (NonEmptyChunk[A], Handle[F, A])]],
	inner: ScopedFuture[F, Pull[F, Nothing, (NonEmptyChunk[B], Handle[F, B])]]
	): Pull[F, B, Nothing] = {
	(outer race inner).pull.flatMap {
	case Left(outer) =>
	outer.optional.flatMap {
	case None =>
	inner.pull.flatMap(identity).flatMap { case (hd, tl) => Pull.output(hd) >> tl.echo }