diff --git a/akka-stream/src/main/scala/akka/stream/impl2/ActorBasedFlowMaterializer.scala b/akka-stream/src/main/scala/akka/stream/impl2/ActorBasedFlowMaterializer.scala
index e925603dbc..db9b03115f 100644
--- a/akka-stream/src/main/scala/akka/stream/impl2/ActorBasedFlowMaterializer.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl2/ActorBasedFlowMaterializer.scala
@@ -13,6 +13,8 @@ import akka.pattern.ask
 import akka.stream.{ MaterializerSettings, Transformer }
 import akka.stream.impl.{ ActorProcessor, ActorPublisher, ExposedPublisher, TransformProcessorImpl }
 import akka.stream.scaladsl2._
+import akka.stream.TimerTransformer
+import akka.stream.impl.TimerTransformerProcessorsImpl
 
 /**
  * INTERNAL API
@@ -24,6 +26,8 @@ private[akka] object Ast {
 
   case class Transform(name: String, mkTransformer: () ⇒ Transformer[Any, Any]) extends AstNode
 
+  case class TimerTransform(name: String, mkTransformer: () ⇒ TimerTransformer[Any, Any]) extends AstNode
+
   case class GroupBy(f: Any ⇒ Any) extends AstNode {
     override def name = "groupBy"
   }
@@ -196,11 +200,11 @@ private[akka] object ActorProcessorFactory {
     val settings = materializer.settings
     (op match {
       case t: Transform      ⇒ Props(new TransformProcessorImpl(settings, t.mkTransformer()))
+      case t: TimerTransform ⇒ Props(new TimerTransformerProcessorsImpl(settings, t.mkTransformer()))
       case g: GroupBy        ⇒ Props(new GroupByProcessorImpl(settings, g.f))
       case tt: PrefixAndTail ⇒ Props(new PrefixAndTailImpl(settings, tt.n))
       case s: SplitWhen      ⇒ Props(new SplitWhenProcessorImpl(settings, s.p))
       case ConcatAll         ⇒ Props(new ConcatAllImpl(materializer))
-
     }).withDispatcher(settings.dispatcher)
   }
 
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl2/Flow.scala b/akka-stream/src/main/scala/akka/stream/scaladsl2/Flow.scala
index cab2f2c6b4..acb495f189 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl2/Flow.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl2/Flow.scala
@@ -3,12 +3,17 @@
  */
 package akka.stream.scaladsl2
 
+import scala.collection.immutable
 import scala.collection.immutable
 import akka.stream.impl2.Ast._
 import org.reactivestreams._
 import scala.annotation.unchecked.uncheckedVariance
 import scala.language.higherKinds
 import akka.stream.Transformer
+import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration.Duration
+import akka.util.Collections.EmptyImmutableSeq
+import akka.stream.TimerTransformer
 
 /**
  * This is the interface from which all concrete Flows inherit. No generic
@@ -17,10 +22,15 @@ import akka.stream.Transformer
  */
 sealed trait Flow
 
+object FlowOps {
+  private case object GroupedWithinTimerKey
+}
+
 /**
  * Operations offered by flows with a free output side: the DSL flows left-to-right only.
  */
 trait FlowOps[-In, +Out] {
+  import FlowOps._
   type Repr[-I, +O] <: FlowOps[I, O]
 
   // Storing ops in reverse order
@@ -53,6 +63,66 @@ trait FlowOps[-In, +Out] {
       override def onNext(in: Out) = if (pf.isDefinedAt(in)) List(pf(in)) else Nil
     })
 
+  /**
+   * Chunk up this stream into groups of the given size, with the last group
+   * possibly smaller than requested due to end-of-stream.
+   *
+   * `n` must be positive, otherwise IllegalArgumentException is thrown.
+   */
+  def grouped(n: Int): Repr[In, immutable.Seq[Out]] = {
+    require(n > 0, "n must be greater than 0")
+    transform("grouped", () ⇒ new Transformer[Out, immutable.Seq[Out]] {
+      var buf: Vector[Out] = Vector.empty
+      override def onNext(in: Out) = {
+        buf :+= in
+        if (buf.size == n) {
+          val group = buf
+          buf = Vector.empty
+          List(group)
+        } else
+          Nil
+      }
+      override def onTermination(e: Option[Throwable]) = if (buf.isEmpty) Nil else List(buf)
+    })
+  }
+
+  /**
+   * Chunk up this stream into groups of elements received within a time window,
+   * or limited by the given number of elements, whatever happens first.
+   * Empty groups will not be emitted if no elements are received from upstream.
+   * The last group before end-of-stream will contain the buffered elements
+   * since the previously emitted group.
+   *
+   * `n` must be positive, and `d` must be greater than 0 seconds, otherwise
+   * IllegalArgumentException is thrown.
+   */
+  def groupedWithin(n: Int, d: FiniteDuration): Repr[In, immutable.Seq[Out]] = {
+    require(n > 0, "n must be greater than 0")
+    require(d > Duration.Zero)
+    timerTransform("groupedWithin", () ⇒ new TimerTransformer[Out, immutable.Seq[Out]] {
+      schedulePeriodically(GroupedWithinTimerKey, d)
+      var buf: Vector[Out] = Vector.empty
+
+      override def onNext(in: Out) = {
+        buf :+= in
+        if (buf.size == n) {
+          // start new time window
+          schedulePeriodically(GroupedWithinTimerKey, d)
+          emitGroup()
+        } else Nil
+      }
+      override def onTermination(e: Option[Throwable]) = if (buf.isEmpty) Nil else List(buf)
+      override def onTimer(timerKey: Any) = emitGroup()
+      private def emitGroup(): immutable.Seq[immutable.Seq[Out]] =
+        if (buf.isEmpty) EmptyImmutableSeq
+        else {
+          val group = buf
+          buf = Vector.empty
+          List(group)
+        }
+    })
+  }
+
   /**
    * Generic transformation of a stream: for each element the [[akka.stream.Transformer#onNext]]
    * function is invoked, expecting a (possibly empty) sequence of output elements
@@ -125,6 +195,33 @@ trait FlowOps[-In, +Out] {
     case _: FlattenStrategy.Concat[Out] ⇒ andThen(ConcatAll)
     case _                              ⇒ throw new IllegalArgumentException(s"Unsupported flattening strategy [${strategy.getClass.getSimpleName}]")
   }
+
+  /**
+   * Transformation of a stream, with additional support for scheduled events.
+   *
+   * For each element the [[akka.stream.Transformer#onNext]]
+   * function is invoked, expecting a (possibly empty) sequence of output elements
+   * to be produced.
+   * After handing off the elements produced from one input element to the downstream
+   * subscribers, the [[akka.stream.Transformer#isComplete]] predicate determines whether to end
+   * stream processing at this point; in that case the upstream subscription is
+   * canceled. Before signaling normal completion to the downstream subscribers,
+   * the [[akka.stream.Transformer#onComplete]] function is invoked to produce a (possibly empty)
+   * sequence of elements in response to the end-of-stream event.
+   *
+   * [[akka.stream.Transformer#onError]] is called when failure is signaled from upstream.
+   *
+   * After normal completion or error the [[akka.stream.Transformer#cleanup]] function is called.
+   *
+   * It is possible to keep state in the concrete [[akka.stream.Transformer]] instance with
+   * ordinary instance variables. The [[akka.stream.Transformer]] is executed by an actor and
+   * therefore you do not have to add any additional thread safety or memory
+   * visibility constructs to access the state from the callback methods.
+   *
+   * Note that you can use [[#transform]] if you just need to transform elements time plays no role in the transformation.
+   */
+  def timerTransform[U](name: String, mkTransformer: () ⇒ TimerTransformer[Out, U]): Repr[In, U] =
+    andThen(TimerTransform(name, mkTransformer.asInstanceOf[() ⇒ TimerTransformer[Any, Any]]))
 }
 
 /**
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedSpec.scala
new file mode 100644
index 0000000000..2b9d715820
--- /dev/null
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedSpec.scala
@@ -0,0 +1,33 @@
+/**
+ * Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+package akka.stream.scaladsl2
+
+import scala.collection.immutable
+import akka.stream.testkit.AkkaSpec
+import akka.stream.testkit2.ScriptedTest
+import scala.concurrent.forkjoin.ThreadLocalRandom.{ current ⇒ random }
+import akka.stream.MaterializerSettings
+
+class FlowGroupedSpec extends AkkaSpec with ScriptedTest {
+
+  val settings = MaterializerSettings(system)
+    .withInputBuffer(initialSize = 2, maxSize = 16)
+    .withFanOutBuffer(initialSize = 1, maxSize = 16)
+
+  "A Grouped" must {
+
+    "group evenly" in {
+      def script = Script((1 to 20) map { _ ⇒ val x, y, z = random.nextInt(); Seq(x, y, z) -> Seq(immutable.Seq(x, y, z)) }: _*)
+      (1 to 30) foreach (_ ⇒ runScript(script, settings)(_.grouped(3)))
+    }
+
+    "group with rest" in {
+      def script = Script(((1 to 20).map { _ ⇒ val x, y, z = random.nextInt(); Seq(x, y, z) -> Seq(immutable.Seq(x, y, z)) }
+        :+ { val x = random.nextInt(); Seq(x) -> Seq(immutable.Seq(x)) }): _*)
+      (1 to 30) foreach (_ ⇒ runScript(script, settings)(_.grouped(3)))
+    }
+
+  }
+
+}
\ No newline at end of file
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedWithinSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedWithinSpec.scala
new file mode 100644
index 0000000000..f9f76e72f2
--- /dev/null
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowGroupedWithinSpec.scala
@@ -0,0 +1,139 @@
+/**
+ * Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+package akka.stream.scaladsl2
+
+import scala.collection.immutable
+import scala.concurrent.duration._
+import scala.concurrent.forkjoin.ThreadLocalRandom.{ current ⇒ random }
+import akka.stream.testkit.AkkaSpec
+import akka.stream.testkit.StreamTestKit
+import akka.stream.testkit2.ScriptedTest
+import akka.stream.MaterializerSettings
+
+@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
+class FlowGroupedWithinSpec extends AkkaSpec with ScriptedTest {
+
+  val settings = MaterializerSettings(system)
+
+  implicit val materializer = FlowMaterializer()
+
+  "A GroupedWithin" must {
+
+    "group elements within the duration" in {
+      val input = Iterator.from(1)
+      val p = StreamTestKit.PublisherProbe[Int]()
+      val c = StreamTestKit.SubscriberProbe[immutable.Seq[Int]]()
+      FlowFrom(p).groupedWithin(1000, 1.second).publishTo(c)
+      val pSub = p.expectSubscription
+      val cSub = c.expectSubscription
+      cSub.request(100)
+      val demand1 = pSub.expectRequest
+      (1 to demand1) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      val demand2 = pSub.expectRequest
+      (1 to demand2) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      val demand3 = pSub.expectRequest
+      c.expectNext((1 to (demand1 + demand2)).toVector)
+      (1 to demand3) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      c.expectNoMsg(300.millis)
+      c.expectNext(((demand1 + demand2 + 1) to (demand1 + demand2 + demand3)).toVector)
+      c.expectNoMsg(300.millis)
+      pSub.expectRequest
+      val last = input.next()
+      pSub.sendNext(last)
+      pSub.sendComplete()
+      c.expectNext(List(last))
+      c.expectComplete
+      c.expectNoMsg(200.millis)
+    }
+
+    "deliver bufferd elements onComplete before the timeout" in {
+      val c = StreamTestKit.SubscriberProbe[immutable.Seq[Int]]()
+      FlowFrom(1 to 3).groupedWithin(1000, 10.second).publishTo(c)
+      val cSub = c.expectSubscription
+      cSub.request(100)
+      c.expectNext((1 to 3).toList)
+      c.expectComplete
+      c.expectNoMsg(200.millis)
+    }
+
+    "buffer groups until requested from downstream" in {
+      val input = Iterator.from(1)
+      val p = StreamTestKit.PublisherProbe[Int]()
+      val c = StreamTestKit.SubscriberProbe[immutable.Seq[Int]]()
+      FlowFrom(p).groupedWithin(1000, 1.second).publishTo(c)
+      val pSub = p.expectSubscription
+      val cSub = c.expectSubscription
+      cSub.request(1)
+      val demand1 = pSub.expectRequest
+      (1 to demand1) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      c.expectNext((1 to demand1).toVector)
+      val demand2 = pSub.expectRequest
+      (1 to demand2) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      c.expectNoMsg(300.millis)
+      cSub.request(1)
+      c.expectNext(((demand1 + 1) to (demand1 + demand2)).toVector)
+      pSub.sendComplete()
+      c.expectComplete
+      c.expectNoMsg(100.millis)
+    }
+
+    "drop empty groups" in {
+      val p = StreamTestKit.PublisherProbe[Int]()
+      val c = StreamTestKit.SubscriberProbe[immutable.Seq[Int]]()
+      FlowFrom(p).groupedWithin(1000, 500.millis).publishTo(c)
+      val pSub = p.expectSubscription
+      val cSub = c.expectSubscription
+      cSub.request(2)
+      pSub.expectRequest
+      c.expectNoMsg(600.millis)
+      pSub.sendNext(1)
+      pSub.sendNext(2)
+      c.expectNext(List(1, 2))
+      // nothing more requested
+      c.expectNoMsg(1100.millis)
+      cSub.request(3)
+      c.expectNoMsg(600.millis)
+      pSub.sendComplete()
+      c.expectComplete
+      c.expectNoMsg(100.millis)
+    }
+
+    "reset time window when max elements reached" in {
+      val input = Iterator.from(1)
+      val p = StreamTestKit.PublisherProbe[Int]()
+      val c = StreamTestKit.SubscriberProbe[immutable.Seq[Int]]()
+      FlowFrom(p).groupedWithin(3, 2.second).publishTo(c)
+      val pSub = p.expectSubscription
+      val cSub = c.expectSubscription
+      cSub.request(4)
+      val demand1 = pSub.expectRequest
+      demand1 should be(4)
+      c.expectNoMsg(1000.millis)
+      (1 to demand1) foreach { _ ⇒ pSub.sendNext(input.next()) }
+      c.probe.within(1000.millis) {
+        c.expectNext((1 to 3).toVector)
+      }
+      c.expectNoMsg(1500.millis)
+      c.probe.within(1000.millis) {
+        c.expectNext(List(4))
+      }
+      pSub.sendComplete()
+      c.expectComplete
+      c.expectNoMsg(100.millis)
+    }
+
+    "group evenly" in {
+      def script = Script((1 to 20) map { _ ⇒ val x, y, z = random.nextInt(); Seq(x, y, z) -> Seq(immutable.Seq(x, y, z)) }: _*)
+      (1 to 30) foreach (_ ⇒ runScript(script, settings)(_.groupedWithin(3, 10.minutes)))
+    }
+
+    "group with rest" in {
+      def script = Script(((1 to 20).map { _ ⇒ val x, y, z = random.nextInt(); Seq(x, y, z) -> Seq(immutable.Seq(x, y, z)) }
+        :+ { val x = random.nextInt(); Seq(x) -> Seq(immutable.Seq(x)) }): _*)
+      (1 to 30) foreach (_ ⇒ runScript(script, settings)(_.groupedWithin(3, 10.minutes)))
+    }
+
+  }
+
+}
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowOnCompleteSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowOnCompleteSpec.scala
index 7528d40cc7..9dc933fc93 100644
--- a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowOnCompleteSpec.scala
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowOnCompleteSpec.scala
@@ -3,7 +3,9 @@
  */
 package akka.stream.scaladsl2
 
-import akka.stream.testkit.{ AkkaSpec, ScriptedTest, StreamTestKit }
+import akka.stream.testkit.{ AkkaSpec, StreamTestKit }
+import akka.stream.testkit.AkkaSpec
+import akka.stream.testkit2.ScriptedTest
 import akka.testkit.TestProbe
 import scala.concurrent.duration._
 import scala.concurrent.forkjoin.ThreadLocalRandom.{ current ⇒ random }
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowPrefixAndTailSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowPrefixAndTailSpec.scala
index 938afc9c3e..e75555fe76 100644
--- a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowPrefixAndTailSpec.scala
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowPrefixAndTailSpec.scala
@@ -11,6 +11,7 @@ import scala.concurrent.Await
 import scala.concurrent.duration._
 import scala.util.control.NoStackTrace
 import akka.stream.MaterializerSettings
+import akka.stream.testkit.StreamTestKit.SubscriberProbe
 
 class FlowPrefixAndTailSpec extends AkkaSpec {
 
@@ -30,14 +31,22 @@ class FlowPrefixAndTailSpec extends AkkaSpec {
       val futureSink = newFutureSink
       val mf = FlowFrom(Nil).prefixAndTail(10).withSink(futureSink).run()
       val fut = futureSink.future(mf)
-      Await.result(fut, 3.seconds) should be((Nil, FlowFrom(EmptyPublisher[Int])))
+      val (prefix, tailFlow) = Await.result(fut, 3.seconds)
+      prefix should be(Nil)
+      val tailSubscriber = SubscriberProbe[Int]
+      tailFlow.publishTo(tailSubscriber)
+      tailSubscriber.expectComplete()
     }
 
     "work on short input" in {
       val futureSink = newFutureSink
       val mf = FlowFrom(List(1, 2, 3)).prefixAndTail(10).withSink(futureSink).run()
       val fut = futureSink.future(mf)
-      Await.result(fut, 3.seconds) should be((List(1, 2, 3), FlowFrom(EmptyPublisher[Int])))
+      val (prefix, tailFlow) = Await.result(fut, 3.seconds)
+      prefix should be(List(1, 2, 3))
+      val tailSubscriber = SubscriberProbe[Int]
+      tailFlow.publishTo(tailSubscriber)
+      tailSubscriber.expectComplete()
     }
 
     "work on longer inputs" in {
@@ -47,11 +56,10 @@ class FlowPrefixAndTailSpec extends AkkaSpec {
       val (takes, tail) = Await.result(fut, 3.seconds)
       takes should be(1 to 5)
 
-      // FIXME enable this again, when grouped is implemented
-      //      val futureSink2 = ???
-      //      val mf2 = tail.grouped(6).withSink(futureSink2).run()
-      //      val fut2 = futureSink2.future(mf2)
-      //      Await.result(fut2, 3.seconds) should be(6 to 10)
+      val futureSink2 = FutureSink[immutable.Seq[Int]]
+      val mf2 = tail.grouped(6).withSink(futureSink2).run()
+      val fut2 = futureSink2.future(mf2)
+      Await.result(fut2, 3.seconds) should be(6 to 10)
     }
 
     "handle zero take count" in {
@@ -61,8 +69,10 @@ class FlowPrefixAndTailSpec extends AkkaSpec {
       val (takes, tail) = Await.result(fut, 3.seconds)
       takes should be(Nil)
 
-      // FIXME enable this again, when grouped is implemented
-      // Await.result(FlowFrom(tail).grouped(11).toFuture(), 3.seconds) should be(1 to 10)
+      val futureSink2 = FutureSink[immutable.Seq[Int]]
+      val mf2 = tail.grouped(11).withSink(futureSink2).run()
+      val fut2 = futureSink2.future(mf2)
+      Await.result(fut2, 3.seconds) should be(1 to 10)
     }
 
     "handle negative take count" in {
@@ -72,8 +82,10 @@ class FlowPrefixAndTailSpec extends AkkaSpec {
       val (takes, tail) = Await.result(fut, 3.seconds)
       takes should be(Nil)
 
-      // FIXME enable this again, when grouped is implemented
-      // Await.result(FlowFrom(tail).grouped(11).toFuture(), 3.seconds) should be(1 to 10)
+      val futureSink2 = FutureSink[immutable.Seq[Int]]
+      val mf2 = tail.grouped(11).withSink(futureSink2).run()
+      val fut2 = futureSink2.future(mf2)
+      Await.result(fut2, 3.seconds) should be(1 to 10)
     }
 
     "work if size of take is equal to stream size" in {
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowToFutureSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowToFutureSpec.scala
index e9da74ef59..b8cfe4034f 100644
--- a/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowToFutureSpec.scala
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/FlowToFutureSpec.scala
@@ -3,7 +3,8 @@
  */
 package akka.stream.scaladsl2
 
-import akka.stream.testkit.{ AkkaSpec, ScriptedTest, StreamTestKit }
+import akka.stream.testkit.{ AkkaSpec, StreamTestKit }
+import akka.stream.testkit2.ScriptedTest
 import scala.concurrent.Await
 import scala.concurrent.duration._
 import scala.concurrent.forkjoin.ThreadLocalRandom.{ current ⇒ random }
diff --git a/akka-stream/src/test/scala/akka/stream/scaladsl2/TimerTransformerSpec.scala b/akka-stream/src/test/scala/akka/stream/scaladsl2/TimerTransformerSpec.scala
new file mode 100644
index 0000000000..8070bd835b
--- /dev/null
+++ b/akka-stream/src/test/scala/akka/stream/scaladsl2/TimerTransformerSpec.scala
@@ -0,0 +1,133 @@
+/**
+ * Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+package akka.stream.scaladsl2
+
+import language.postfixOps
+import scala.collection.immutable
+import scala.concurrent.duration._
+import akka.actor.Actor
+import akka.actor.ActorCell
+import akka.actor.ActorRef
+import akka.actor.Props
+import akka.stream.TimerTransformer.Scheduled
+import akka.stream.testkit.AkkaSpec
+import akka.testkit.TestDuration
+import akka.testkit.TestKit
+import akka.stream.TimerTransformer
+
+object TimerTransformerSpec {
+  case object TestSingleTimer
+  case object TestSingleTimerResubmit
+  case object TestCancelTimer
+  case object TestCancelTimerAck
+  case object TestRepeatedTimer
+  case class Tick(n: Int)
+
+  def driverProps(probe: ActorRef): Props =
+    Props(classOf[Driver], probe).withDispatcher("akka.test.stream-dispatcher")
+
+  class Driver(probe: ActorRef) extends Actor {
+
+    // need implicit system for dilated
+    import context.system
+
+    val tickCount = Iterator from 1
+
+    val transformer = new TimerTransformer[Int, Int] {
+      override def onNext(elem: Int): immutable.Seq[Int] = List(elem)
+      override def onTimer(timerKey: Any): immutable.Seq[Int] = {
+        val tick = Tick(tickCount.next())
+        probe ! tick
+        if (timerKey == "TestSingleTimerResubmit" && tick.n == 1)
+          scheduleOnce("TestSingleTimerResubmit", 500.millis.dilated)
+        else if (timerKey == "TestRepeatedTimer" && tick.n == 5)
+          cancelTimer("TestRepeatedTimer")
+        Nil
+      }
+    }
+
+    override def preStart(): Unit = {
+      super.preStart()
+      transformer.start(context)
+    }
+
+    override def postStop(): Unit = {
+      super.postStop()
+      transformer.stop()
+    }
+
+    def receive = {
+      case TestSingleTimer ⇒
+        transformer.scheduleOnce("TestSingleTimer", 500.millis.dilated)
+      case TestSingleTimerResubmit ⇒
+        transformer.scheduleOnce("TestSingleTimerResubmit", 500.millis.dilated)
+      case TestCancelTimer ⇒
+        transformer.scheduleOnce("TestCancelTimer", 1.milli.dilated)
+        TestKit.awaitCond(context.asInstanceOf[ActorCell].mailbox.hasMessages, 1.second.dilated)
+        transformer.cancelTimer("TestCancelTimer")
+        probe ! TestCancelTimerAck
+        transformer.scheduleOnce("TestCancelTimer", 500.milli.dilated)
+      case TestRepeatedTimer ⇒
+        transformer.schedulePeriodically("TestRepeatedTimer", 100.millis.dilated)
+      case s: Scheduled ⇒ transformer.onScheduled(s)
+    }
+  }
+}
+
+@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
+class TimerTransformerSpec extends AkkaSpec {
+  import TimerTransformerSpec._
+
+  "A TimerTransformer" must {
+
+    "receive single-shot timer" in {
+      val driver = system.actorOf(driverProps(testActor))
+      within(2 seconds) {
+        within(500 millis, 1 second) {
+          driver ! TestSingleTimer
+          expectMsg(Tick(1))
+        }
+        expectNoMsg(1 second)
+      }
+    }
+
+    "resubmit single-shot timer" in {
+      val driver = system.actorOf(driverProps(testActor))
+      within(2.5 seconds) {
+        within(500 millis, 1 second) {
+          driver ! TestSingleTimerResubmit
+          expectMsg(Tick(1))
+        }
+        within(1 second) {
+          expectMsg(Tick(2))
+        }
+        expectNoMsg(1 second)
+      }
+    }
+
+    "correctly cancel a named timer" in {
+      val driver = system.actorOf(driverProps(testActor))
+      driver ! TestCancelTimer
+      within(500 millis) {
+        expectMsg(TestCancelTimerAck)
+      }
+      within(300 millis, 1 second) {
+        expectMsg(Tick(1))
+      }
+      expectNoMsg(1 second)
+    }
+
+    "receive and cancel a repeated timer" in {
+      val driver = system.actorOf(driverProps(testActor))
+      driver ! TestRepeatedTimer
+      val seq = receiveWhile(2 seconds) {
+        case t: Tick ⇒ t
+      }
+      seq should have length 5
+      expectNoMsg(1 second)
+    }
+
+  }
+
+}