/*
 * Copyright (C) 2014-2018 Lightbend Inc. <https://www.lightbend.com>
 */

package akka.stream.scaladsl

import scala.concurrent.duration._
import akka.stream.ActorMaterializer
import akka.stream.testkit._
import akka.stream.testkit.Utils._
import akka.stream.testkit.scaladsl.StreamTestKit._

class FlowTakeWithinSpec extends StreamSpec {

  implicit val materializer = ActorMaterializer()

  "A TakeWithin" must {

    "deliver elements within the duration, but not afterwards" in {
      val input = Iterator.from(1)
      val p = TestPublisher.manualProbe[Int]()
      val c = TestSubscriber.manualProbe[Int]()
      Source.fromPublisher(p).takeWithin(1.second).to(Sink.fromSubscriber(c)).run()
      val pSub = p.expectSubscription()
      val cSub = c.expectSubscription()
      cSub.request(100)
      val demand1 = pSub.expectRequest().toInt
      (1 to demand1) foreach { _ ⇒ pSub.sendNext(input.next()) }
      val demand2 = pSub.expectRequest().toInt
      (1 to demand2) foreach { _ ⇒ pSub.sendNext(input.next()) }
      val demand3 = pSub.expectRequest().toInt
      val sentN = demand1 + demand2
      (1 to sentN) foreach { n ⇒ c.expectNext(n) }
      within(2.seconds) {
        c.expectComplete()
      }
      (1 to demand3) foreach { _ ⇒ pSub.sendNext(input.next()) }
      c.expectNoMsg(200.millis)
    }

    "deliver buffered elements onComplete before the timeout" in assertAllStagesStopped {
      val c = TestSubscriber.manualProbe[Int]()
      Source(1 to 3).takeWithin(1.second).to(Sink.fromSubscriber(c)).run()
      val cSub = c.expectSubscription()
      c.expectNoMsg(200.millis)
      cSub.request(100)
      (1 to 3) foreach { n ⇒ c.expectNext(n) }
      c.expectComplete()
      c.expectNoMsg(200.millis)
    }

  }

}
