raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Merge pull request #18260 from akka/wip-15040-ScatterGatherFirstCompleted-patriknw

Browse source 
=act #15040 ScatterGatherFirstCompleted reply when no routees
This commit is contained in:
Patrik Nordwall 2015-08-18 20:00:37 +02:00
commit 1732e62d0c
2 changed files with 35 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

14
akka-actor-tests/src/test/scala/akka/routing/ScatterGatherFirstCompletedSpec.scala
View file

@ -10,6 +10,9 @@ import akka.actor.{ Props, Actor }
import akka.pattern.ask import akka.pattern.ask
import akka.testkit.{ TestLatch, ImplicitSender, DefaultTimeout, AkkaSpec } import akka.testkit.{ TestLatch, ImplicitSender, DefaultTimeout, AkkaSpec }
import akka.actor.ActorSystem import akka.actor.ActorSystem
import akka.actor.Status
import java.util.concurrent.TimeoutException
import akka.testkit.TestProbe
object ScatterGatherFirstCompletedSpec { object ScatterGatherFirstCompletedSpec {
class TestActor extends Actor { class TestActor extends Actor {
@ -86,4 +89,15 @@ class ScatterGatherFirstCompletedSpec extends AkkaSpec with DefaultTimeout with
} }
"Scatter-gather pool" must {
"without routees should reply immediately" in {
val probe = TestProbe()
val router = system.actorOf(ScatterGatherFirstCompletedPool(nrOfInstances = 0, within = 5.seconds).props(Props.empty))
router.tell("hello", probe.ref)
probe.expectMsgType[Status.Failure](2.seconds).cause.getClass should be(classOf[TimeoutException])
}
}
} }

36
akka-actor/src/main/scala/akka/routing/ScatterGatherFirstCompleted.scala
View file

@ -21,6 +21,8 @@ import akka.util.Timeout
import akka.util.Helpers.ConfigOps import akka.util.Helpers.ConfigOps
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import akka.actor.ActorSystem import akka.actor.ActorSystem
import scala.concurrent.Future
import java.util.concurrent.TimeoutException
/** /**
* Broadcasts the message to all routees, and replies with the first response. * Broadcasts the message to all routees, and replies with the first response.
@ -31,8 +33,7 @@ import akka.actor.ActorSystem
@SerialVersionUID(1L) @SerialVersionUID(1L)
final case class ScatterGatherFirstCompletedRoutingLogic(within: FiniteDuration) extends RoutingLogic { final case class ScatterGatherFirstCompletedRoutingLogic(within: FiniteDuration) extends RoutingLogic {
override def select(message: Any, routees: immutable.IndexedSeq[Routee]): Routee = override def select(message: Any, routees: immutable.IndexedSeq[Routee]): Routee =
if (routees.isEmpty) NoRoutee ScatterGatherFirstCompletedRoutees(routees, within)
else ScatterGatherFirstCompletedRoutees(routees, within)
} }
/** /**
@ -42,20 +43,25 @@ final case class ScatterGatherFirstCompletedRoutingLogic(within: FiniteDuration)
private[akka] final case class ScatterGatherFirstCompletedRoutees( private[akka] final case class ScatterGatherFirstCompletedRoutees(
routees: immutable.IndexedSeq[Routee], within: FiniteDuration) extends Routee { routees: immutable.IndexedSeq[Routee], within: FiniteDuration) extends Routee {
override def send(message: Any, sender: ActorRef): Unit = { override def send(message: Any, sender: ActorRef): Unit =
implicit val ec = ExecutionContexts.sameThreadExecutionContext if (routees.isEmpty) {
implicit val timeout = Timeout(within) implicit val ec = ExecutionContexts.sameThreadExecutionContext
val promise = Promise[Any]() val reply = Future.failed(new TimeoutException("Timeout due to no routees"))
routees.foreach { reply.pipeTo(sender)
case ActorRefRoutee(ref) ⇒ } else {
promise.tryCompleteWith(ref.ask(message)) implicit val ec = ExecutionContexts.sameThreadExecutionContext
case ActorSelectionRoutee(sel) ⇒ implicit val timeout = Timeout(within)
promise.tryCompleteWith(sel.ask(message)) val promise = Promise[Any]()
case _ ⇒ routees.foreach {
} case ActorRefRoutee(ref) ⇒
promise.tryCompleteWith(ref.ask(message))
case ActorSelectionRoutee(sel) ⇒
promise.tryCompleteWith(sel.ask(message))
case _ ⇒
}
promise.future.pipeTo(sender) promise.future.pipeTo(sender)
} }
} }
/** /**