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

stop ClusterClient ResponseTunnel after first reply when ask is used, #22093

Browse source
This commit is contained in:
Patrik Nordwall 2017-01-03 16:42:31 +01:00
parent d1e255f42f
commit b213a7af20
2 changed files with 33 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

11
akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala
View file

@ -812,12 +812,21 @@ object ClusterReceptionist {
*/ */
class ClientResponseTunnel(client: ActorRef, timeout: FiniteDuration) extends Actor with ActorLogging { class ClientResponseTunnel(client: ActorRef, timeout: FiniteDuration) extends Actor with ActorLogging {
context.setReceiveTimeout(timeout) context.setReceiveTimeout(timeout)
private val isAsk = {
val pathElements = client.path.elements
pathElements.size == 2 && pathElements.head == "temp" && pathElements.tail.head.startsWith("$")
}
def receive = { def receive = {
case Ping ⇒ // keep alive from client case Ping ⇒ // keep alive from client
case ReceiveTimeout ⇒ case ReceiveTimeout ⇒
log.debug("ClientResponseTunnel for client [{}] stopped due to inactivity", client.path) log.debug("ClientResponseTunnel for client [{}] stopped due to inactivity", client.path)
context stop self context stop self
case msg ⇒ client.tell(msg, Actor.noSender) case msg ⇒
client.tell(msg, Actor.noSender)
if (isAsk)
context stop self
} }
} }
} }

29
akka-cluster-tools/src/multi-jvm/scala/akka/cluster/client/ClusterClientSpec.scala
View file

@ -213,6 +213,24 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
enterBarrier("after-2") enterBarrier("after-2")
} }
"work with ask" in within(10 seconds) {
runOn(client) {
import akka.pattern.ask
import system.dispatcher
val c = system.actorOf(ClusterClient.props(
ClusterClientSettings(system).withInitialContacts(initialContacts)), "ask-client")
implicit val timeout = Timeout(remaining)
val reply = c ? ClusterClient.Send("/user/testService", "hello-request", localAffinity = true)
Await.result(reply.mapTo[Reply], remaining).msg should be("hello-request-ack")
system.stop(c)
}
runOn(fourth) {
expectMsg("hello-request")
}
enterBarrier("after-3")
}
"demonstrate usage" in within(15 seconds) { "demonstrate usage" in within(15 seconds) {
def host1 = first def host1 = first
def host2 = second def host2 = second
@ -261,7 +279,7 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
// strange, barriers fail without this sleep // strange, barriers fail without this sleep
Thread.sleep(1000) Thread.sleep(1000)
enterBarrier("after-3") enterBarrier("after-4")
} }
"report events" in within(15 seconds) { "report events" in within(15 seconds) {
@ -305,7 +323,7 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
} }
} }
enterBarrier("after-6") enterBarrier("after-5")
} }
"re-establish connection to another receptionist when server is shutdown" in within(30 seconds) { "re-establish connection to another receptionist when server is shutdown" in within(30 seconds) {
@ -356,7 +374,7 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
} }
} }
} }
enterBarrier("after-4") enterBarrier("after-6")
} }
"re-establish connection to receptionist after partition" in within(30 seconds) { "re-establish connection to receptionist after partition" in within(30 seconds) {
@ -397,7 +415,7 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
system.stop(c) system.stop(c)
} }
enterBarrier("after-5") enterBarrier("after-7")
} }
"re-establish connection to receptionist after server restart" in within(30 seconds) { "re-establish connection to receptionist after server restart" in within(30 seconds) {
@ -436,8 +454,7 @@ class ClusterClientSpec extends MultiNodeSpec(ClusterClientSpec) with STMultiNod
system.name, system.name,
ConfigFactory.parseString( ConfigFactory.parseString(
if (RARP(system).provider.remoteSettings.Artery.Enabled) s"akka.remote.artery.canonical.port=$port" if (RARP(system).provider.remoteSettings.Artery.Enabled) s"akka.remote.artery.canonical.port=$port"
else s"akka.remote.netty.tcp.port=$port" else s"akka.remote.netty.tcp.port=$port").withFallback(system.settings.config))
).withFallback(system.settings.config))
Cluster(sys2).join(Cluster(sys2).selfAddress) Cluster(sys2).join(Cluster(sys2).selfAddress)
val service2 = sys2.actorOf(Props(classOf[TestService], testActor), "service2") val service2 = sys2.actorOf(Props(classOf[TestService], testActor), "service2")
ClusterClientReceptionist(sys2).registerService(service2) ClusterClientReceptionist(sys2).registerService(service2)