BufferIntrospectiveTest.scala

     val scheduler = schedulerProvider.defaultScheduler

      val xs = collection.mutable.ArrayBuffer[Seq[Long]]()
      var completed = false

      val o =
        Observable
          .intervalAtFixedRate(timeSlice)
          .take(10)

      o.bufferIntrospective(scheduler)
        .subscribe(v ⇒ {
          schedulerProvider.advanceTimeBy(timeSlice * 3 + 1.micro)
          xs += v
          Ack.Continue
        },
          e ⇒ e.printStackTrace(),
          () ⇒ completed = true
        )(schedulerProvider.defaultScheduler)

      schedulerProvider.advanceTimeBy(timeSlice * 15)
      assert(completed)
      val xsList = xs.map(_.toList)
      assert(xsList.size == 4)
      assert(xsList.head == List(0))
      assert(xsList(1) == List(1, 2, 3))
      assert(xsList(2) == List(4, 5, 6))
      assert(xsList(3) == List(7, 8, 9))