Non terminating scalaz streams zipwith.

ZipWith

import concurrent.duration._

import scalaz.stream._
import scalaz.concurrent._

import org.joda.time.DateTime

/**
 *  Requires the following libs :
 * 
 *  "org.scalaz.stream"         %%  "scalaz-stream"     % "0.5a",
 *  "joda-time"                 %   "joda-time"         % "2.4"
 * 
 */ 

object Noodle extends App {

  implicit val mooose = Strategy.DefaultTimeoutScheduler

  val input = Process.emitAll(1 to 5)

  val pulse = Process.awakeEvery(1.second)

  val console: Sink[Task, Int] = Process.constant { case i ⇒ Task.delay(println(s"$i ${DateTime.now()}")) }

  val ignorePulse = { (i: Int, _: Duration) ⇒ i }

  input.zipWith(pulse)(ignorePulse)
    .observe(console)
    .run
    .run
}