scalaz-stream pipeIn

gistfile1.scala

// doesn't print "release" unless you remove `.pipeIn(process1.id[A])`,
// internally `pipeIn` uses `takeWhile` which uses `pipe`, it breaks print too

def printSink[A] = io.stdOutLines.contramap[A](_.toString).pipeIn(process1.id[A])
// simplier case doesn't work too:
//   sink.lift[Task, A](x => Task.delay(())).pipe(process1.id[A => Task[Unit]])

def disposableThing: Process[Task, Unit] = {
  val acquire = Task.delay {println("acquire")}
  val release = Task.delay {println("release")}

  io.resource(acquire)(x => release)(Task.now)
}

val poison = async.signalOf[Boolean](false)

new Thread {
  override def run(): Unit = {
    Thread.sleep(3000)
    poison.set(true).run
  }
}.start()

poison.discrete.wye(
  for {
    thing <- disposableThing
    y <- Process.emit(1).observe(printSink)
  } yield y
)(wye.interrupt).run.run