This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 3b21abca49 fix(remote): resume classic remoting reader after failed
handoff (#3205)
3b21abca49 is described below
commit 3b21abca495d00d53229f182ba510b28ca4101f3
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 5 21:17:27 2026 +0800
fix(remote): resume classic remoting reader after failed handoff (#3205)
Motivation:
Classic remoting passive read handoff can leave the original outbound
reader stuck in notReading if the read-only handoff endpoint fails.
Modification:
Track passive read handoff fallback endpoints and resume the original
writable reader when the read-only endpoint terminates while the original
endpoint and UID still match. Carry fallback state across replacement read-only
endpoints, clear stale fallback state on UID/address changes, handle
ResumeReading in supervisor states, add logging, and add regression coverage.
Also correct the stale classic remoting passive-connections config text.
Result:
Classic remoting can recover future inbound messages on the original
association after a failed passive read handoff.
Tests:
- rtk sbt "remote / Test / testOnly
org.apache.pekko.remote.ReliableDeliverySupervisorSpec" / passed (4 tests)
- rtk sbt "remote / Test / testOnly
org.apache.pekko.remote.classic.RemotingSpec -- -z \"resume the outbound
reader\"" / passed
- rtk sbt "remote / Test / testOnly
org.apache.pekko.remote.classic.RemotingSpec" / passed (22 tests)
- rtk scalafmt --list --mode diff-ref=origin/main / passed (JVM Unsafe
warnings only)
- rtk bash -lc 'git diff --check' / passed
- rtk sbt sortImports / failed: Scalafix/scala-meta NoSuchMethodError in
local tooling; unrelated generated change was reverted
- qodercli stdout review / No must-fix findings
- independent subAgent review / No must-fix findings
References:
Fixes #3206
---
.../scala/org/apache/pekko/remote/Endpoint.scala | 46 +++++++--
.../scala/org/apache/pekko/remote/Remoting.scala | 85 ++++++++++++++--
.../remote/ReliableDeliverySupervisorSpec.scala | 101 ++++++++++++++++---
.../apache/pekko/remote/classic/RemotingSpec.scala | 111 +++++++++++++++++++++
4 files changed, 313 insertions(+), 30 deletions(-)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
index 0c39a1e4de..8e0208bade 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
@@ -387,6 +387,9 @@ private[remote] class ReliableDeliverySupervisor(
case s: EndpointWriter.StopReading =>
writer.forward(s)
+ case EndpointWriter.ResumeReading =>
+ writer ! EndpointWriter.ResumeReading
+
case Ungate => // ok, not gated
}
@@ -418,10 +421,18 @@ private[remote] class ReliableDeliverySupervisor(
// Resending will be triggered by the incoming GotUid message after
the connection finished
goToActive()
} else goToIdle()
- case AttemptSysMsgRedelivery => // Ignore
- case s @ Send(_: SystemMessage, _, _, _) => tryBuffer(s.copy(seqOpt =
Some(nextSeq())))
- case s: Send => context.system.deadLetters
! s
- case EndpointWriter.FlushAndStop => context.stop(self)
+ case AttemptSysMsgRedelivery => // Ignore
+ case s @ Send(_: SystemMessage, _, _, _) => tryBuffer(s.copy(seqOpt =
Some(nextSeq())))
+ case s: Send => context.system.deadLetters ! s
+ case EndpointWriter.FlushAndStop => context.stop(self)
+ case EndpointWriter.ResumeReading =>
+ if (!writerTerminated) {
+ writer ! EndpointWriter.ResumeReading
+ context.become(gated(writerTerminated = false, earlyUngateRequested =
true))
+ } else {
+ writer = createWriter()
+ goToActive()
+ }
case EndpointWriter.StopReading(w, replyTo) =>
replyTo ! EndpointWriter.StoppedReading(w)
sender() ! EndpointWriter.StoppedReading(w)
@@ -448,7 +459,10 @@ private[remote] class ReliableDeliverySupervisor(
new TimeoutException(
"Remote system has been silent for too long. " +
s"(more than ${settings.QuarantineSilentSystemTimeout.toCoarsest})"))
- case EndpointWriter.FlushAndStop => context.stop(self)
+ case EndpointWriter.FlushAndStop => context.stop(self)
+ case EndpointWriter.ResumeReading =>
+ writer = createWriter()
+ goToActive()
case EndpointWriter.StopReading(w, replyTo) =>
replyTo ! EndpointWriter.StoppedReading(w)
case Ungate => // ok, not gated
@@ -482,7 +496,8 @@ private[remote] class ReliableDeliverySupervisor(
// and don't really know if they were properly delivered or not.
resendBuffer = new AckedSendBuffer[Send](0)
context.stop(self)
- case _ => // Ignore
+ case EndpointWriter.ResumeReading => // Ignore while shutting down
+ case _ => // Ignore
}
private def handleSend(send: Send): Unit =
@@ -625,6 +640,7 @@ private[remote] object EndpointWriter {
private case object FlushAndStopTimeout
case object AckIdleCheckTimer
final case class StopReading(writer: ActorRef, replyTo: ActorRef)
+ case object ResumeReading
final case class StoppedReading(writer: ActorRef)
final case class Handle(handle: PekkoProtocolHandle) extends
NoSerializationVerificationNeeded
@@ -751,8 +767,10 @@ private[remote] class EndpointWriter(
}
val buffering: Receive = {
- case s: Send => enqueueInBuffer(s)
- case BackoffTimer => sendBufferedMessages()
+ case s: Send => enqueueInBuffer(s)
+ case BackoffTimer => sendBufferedMessages()
+ case ResumeReading =>
+ reader.foreach(_ ! ResumeReading)
case FlushAndStop =>
// Flushing is postponed after the pending writes
buffer.offer(FlushAndStop)
@@ -804,6 +822,9 @@ private[remote] class EndpointWriter(
case s @ StopReading(_, replyTo) =>
reader.foreach(_.tell(s, replyTo))
true
+ case ResumeReading =>
+ reader.foreach(_ ! ResumeReading)
+ true
case unexpected =>
throw new IllegalArgumentException(s"Unexpected message type:
${unexpected.getClass}")
}
@@ -996,6 +1017,8 @@ private[remote] class EndpointWriter(
// initializing, buffer and take care of it later when buffer is sent
enqueueInBuffer(s)
}
+ case ResumeReading =>
+ reader.foreach(_ ! ResumeReading)
case TakeOver(newHandle, replyTo) =>
// Shutdown old reader
handle.foreach { _.disassociate("the association was replaced by a new
one", log) }
@@ -1113,7 +1136,7 @@ private[remote] class EndpointReader(
val receiveBuffers: ConcurrentHashMap[Link, ResendState])
extends EndpointActor(localAddress, remoteAddress, transport, settings,
codec) {
- import EndpointWriter.{ OutboundAck, StopReading, StoppedReading }
+ import EndpointWriter.{ OutboundAck, ResumeReading, StopReading,
StoppedReading }
val provider = RARP(context.system).provider
var ackedReceiveBuffer = new AckedReceiveBuffer[Message]
@@ -1186,6 +1209,8 @@ private[remote] class EndpointReader(
context.become(notReading)
replyTo ! StoppedReading(writer)
+ case ResumeReading => // already reading
+
}
private def logTransientSerializationError(msg: PekkoPduCodec.Message,
error: Throwable): Unit = {
@@ -1204,6 +1229,9 @@ private[remote] class EndpointReader(
case StopReading(writer, replyTo) =>
replyTo ! StoppedReading(writer)
+ case ResumeReading =>
+ context.become(receive)
+
case InboundPayload(p) if p.length <= transport.maximumPayloadBytes =>
val (ackOption, msgOption) = tryDecodeMessageAndAck(p)
for (ack <- ackOption; reliableDelivery <- reliableDeliverySupervisor)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
index 2aca007491..3aed2fcc75 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
@@ -373,6 +373,7 @@ private[remote] object EndpointManager {
final case class Quarantined(uid: Int, timeOfRelease: Deadline) extends
EndpointPolicy {
override def isTombstone: Boolean = true
}
+ private final case class ResumableReader(writer: ActorRef, remoteAddress:
Address, uid: Int)
// Not threadsafe -- only to be used in HeadActor
class EndpointRegistry {
@@ -536,6 +537,7 @@ private[remote] class EndpointManager(conf: Config, log:
LoggingAdapter)
context.system.scheduler.scheduleWithFixedDelay(pruneInterval,
pruneInterval, self, Prune)
var pendingReadHandoffs = Map[ActorRef, PekkoProtocolHandle]()
+ private var readOnlyReaderResumptions = Map[ActorRef, ResumableReader]()
var stashedInbound = Map[ActorRef, Vector[InboundAssociation]]()
def handleStashedInbound(endpoint: ActorRef, writerIsIdle: Boolean): Unit = {
@@ -783,13 +785,17 @@ private[remote] class EndpointManager(conf: Config, log:
LoggingAdapter)
case ia @ InboundAssociation(_: PekkoProtocolHandle) =>
handleInboundAssociation(ia, writerIsIdle = false)
case EndpointWriter.StoppedReading(endpoint) =>
- acceptPendingReader(takingOverFrom = endpoint)
+ acceptPendingReader(takingOverFrom = endpoint, resumeReadingFrom =
Some(endpoint))
case Terminated(endpoint) =>
- acceptPendingReader(takingOverFrom = endpoint)
- endpoints.unregisterEndpoint(endpoint)
+ if (pendingReadHandoffs.contains(endpoint))
+ endpoints.unregisterEndpoint(endpoint)
+ if (!acceptPendingReader(takingOverFrom = endpoint, resumeReadingFrom =
None)) {
+ resumeReadingIfNeeded(readOnlyEndpoint = endpoint)
+ endpoints.unregisterEndpoint(endpoint)
+ }
handleStashedInbound(endpoint, writerIsIdle = false)
case EndpointWriter.TookOver(endpoint, handle) =>
- removePendingReader(takingOverFrom = endpoint, withHandle = handle)
+ completePendingReaderTakeOver(takingOverFrom = endpoint, withHandle =
handle)
case ReliableDeliverySupervisor.GotUid(uid, remoteAddress) =>
val refuseUidOption = endpoints.refuseUid(remoteAddress)
endpoints.writableEndpointWithPolicyFor(remoteAddress) match {
@@ -959,7 +965,7 @@ private[remote] class EndpointManager(conf: Config, log:
LoggingAdapter)
})
}
- private def acceptPendingReader(takingOverFrom: ActorRef): Unit = {
+ private def acceptPendingReader(takingOverFrom: ActorRef, resumeReadingFrom:
Option[ActorRef]): Boolean = {
if (pendingReadHandoffs.contains(takingOverFrom)) {
val handle = pendingReadHandoffs(takingOverFrom)
pendingReadHandoffs -= takingOverFrom
@@ -973,12 +979,77 @@ private[remote] class EndpointManager(conf: Config, log:
LoggingAdapter)
Some(handle),
writing = false)
endpoints.registerReadOnlyEndpoint(handle.remoteAddress, endpoint,
handle.handshakeInfo.uid)
+ val inheritedResumableReader =
readOnlyReaderResumptions.get(takingOverFrom).filter(
+ resumableReaderMatchesHandle(_, readOnlyEndpoint = takingOverFrom,
handle = handle))
+ val resumableReader = resumeReadingFrom
+ .map(writer => ResumableReader(writer, handle.remoteAddress,
handle.handshakeInfo.uid))
+ .orElse(inheritedResumableReader)
+ readOnlyReaderResumptions -= takingOverFrom
+ resumableReader.foreach {
+ case ResumableReader(writer, remoteAddress, uid) =>
+ readOnlyReaderResumptions += endpoint -> ResumableReader(writer,
remoteAddress, uid)
+ log.debug(
+ "Registered passive read handoff fallback for [{}] with UID [{}]
from writable endpoint [{}] to read-only endpoint [{}]",
+ remoteAddress,
+ uid,
+ writer,
+ endpoint)
+ }
+ true
+ } else false
+ }
+
+ private def resumeReadingIfNeeded(readOnlyEndpoint: ActorRef): Unit = {
+ readOnlyReaderResumptions.get(readOnlyEndpoint).foreach {
+ case ResumableReader(writer, remoteAddress, uid) =>
+ endpoints.writableEndpointWithPolicyFor(remoteAddress) match {
+ case Some(Pass(`writer`, Some(`uid`))) =>
+ log.info(
+ "Resuming outbound reader [{}] for [{}] with UID [{}] after
passive read-only endpoint [{}] stopped",
+ writer,
+ remoteAddress,
+ uid,
+ readOnlyEndpoint)
+ writer ! EndpointWriter.ResumeReading
+ case currentPolicy =>
+ log.debug(
+ "Not resuming outbound reader [{}] for [{}] with UID [{}] after
passive read-only endpoint [{}] stopped",
+ writer,
+ remoteAddress,
+ uid,
+ readOnlyEndpoint)
+ log.debug(
+ "Current endpoint policy when skipping passive read handoff
fallback for [{}] is [{}]",
+ remoteAddress,
+ currentPolicy)
+ }
}
+ readOnlyReaderResumptions -= readOnlyEndpoint
}
- private def removePendingReader(takingOverFrom: ActorRef, withHandle:
PekkoProtocolHandle): Unit = {
- if (pendingReadHandoffs.get(takingOverFrom).exists(handle => handle ==
withHandle))
+ private def resumableReaderMatchesHandle(
+ resumableReader: ResumableReader,
+ readOnlyEndpoint: ActorRef,
+ handle: PekkoProtocolHandle): Boolean = {
+ val ResumableReader(writer, remoteAddress, uid) = resumableReader
+ val matchesCurrentHandle = remoteAddress == handle.remoteAddress && uid ==
handle.handshakeInfo.uid
+ if (!matchesCurrentHandle && log.isDebugEnabled)
+ log.debug(
+ s"""Dropping passive read handoff fallback for [$remoteAddress] with
UID [$uid] from writable endpoint
+ |[$writer] because read-only endpoint [$readOnlyEndpoint] now
handles [${handle.remoteAddress}]
+ |with UID [${handle.handshakeInfo.uid}]""".stripMargin)
+ matchesCurrentHandle
+ }
+
+ private def completePendingReaderTakeOver(takingOverFrom: ActorRef,
withHandle: PekkoProtocolHandle): Unit = {
+ if (pendingReadHandoffs.get(takingOverFrom).exists(handle => handle ==
withHandle)) {
pendingReadHandoffs -= takingOverFrom
+ endpoints.registerReadOnlyEndpoint(withHandle.remoteAddress,
takingOverFrom, withHandle.handshakeInfo.uid)
+ if (!readOnlyReaderResumptions
+ .get(takingOverFrom)
+ .forall(resumableReaderMatchesHandle(_, readOnlyEndpoint =
takingOverFrom, handle = withHandle)))
+ readOnlyReaderResumptions -= takingOverFrom
+ }
}
private def createEndpoint(
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/ReliableDeliverySupervisorSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/ReliableDeliverySupervisorSpec.scala
index d433f03bc0..4529eb21fe 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/ReliableDeliverySupervisorSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/ReliableDeliverySupervisorSpec.scala
@@ -21,9 +21,10 @@ import java.util.concurrent.ConcurrentHashMap
import scala.annotation.nowarn
import scala.concurrent.Promise
+import scala.concurrent.duration._
import org.apache.pekko
-import pekko.actor.{ ActorRef, Address, Nobody, RootActorPath }
+import pekko.actor.{ ActorRef, Address, Nobody, RootActorPath, Terminated }
import pekko.remote.EndpointManager.{ Link, ResendState, Send }
import pekko.remote.ReliableDeliverySupervisor.AckFromReader
import pekko.remote.transport._
@@ -90,6 +91,61 @@ class ReliableDeliverySupervisorSpec extends
PekkoSpec(ReliableDeliverySuperviso
underlying.resendBuffer.nonAcked.map(_.seq) should ===(Vector(SeqNo(1)))
underlying.resendBuffer = new AckedSendBuffer[Send](0)
}
+
+ "recreate the writer when resuming reading while idle" in {
+ val parentProbe = TestProbe()
+ val supervisor = newSupervisor(initialUid = oldUid, parentProbe.ref)
+ val oldWriter = supervisor.underlyingActor.writer
+ val writerProbe = TestProbe()
+
+ writerProbe.watch(oldWriter)
+ supervisor.unwatch(oldWriter)
+ system.stop(oldWriter)
+ writerProbe.expectTerminated(oldWriter)
+
+ supervisor.receive(Terminated(oldWriter)(existenceConfirmed = true,
addressTerminated = false))
+ parentProbe.expectMsg(EndpointWriter.StoppedReading(supervisor))
+ supervisor.underlyingActor.currentHandle =
Some(newProtocolHandle(oldUid))
+
+ supervisor.receive(EndpointWriter.ResumeReading)
+
+ supervisor.underlyingActor.writer should not be oldWriter
+ expectActive(supervisor)
+ }
+
+ "recreate the writer when resuming reading while gated" in {
+ val parentProbe = TestProbe()
+ val supervisor = newSupervisor(initialUid = oldUid, parentProbe.ref)
+ val underlying = supervisor.underlyingActor
+ val oldWriter = supervisor.underlyingActor.writer
+ val writerProbe = TestProbe()
+
+ writerProbe.watch(oldWriter)
+ supervisor.unwatch(oldWriter)
+ system.stop(oldWriter)
+ writerProbe.expectTerminated(oldWriter)
+ underlying.currentHandle = Some(newProtocolHandle(oldUid))
+ underlying.context.become(underlying.gated(writerTerminated = true,
earlyUngateRequested = false))
+
+ supervisor.receive(EndpointWriter.ResumeReading)
+
+ supervisor.underlyingActor.writer should not be oldWriter
+ expectActive(supervisor)
+ }
+
+ "forward resume reading while gated before the writer terminates" in {
+ val parentProbe = TestProbe()
+ val supervisor = newSupervisor(initialUid = oldUid, parentProbe.ref)
+ val underlying = supervisor.underlyingActor
+ val writerProbe = TestProbe()
+
+ underlying.writer = writerProbe.ref
+ underlying.context.become(underlying.gated(writerTerminated = false,
earlyUngateRequested = false))
+
+ supervisor.receive(EndpointWriter.ResumeReading)
+
+ writerProbe.expectMsg(EndpointWriter.ResumeReading)
+ }
}
private def newSupervisor(initialUid: Int, parent: ActorRef):
TestActorRef[ReliableDeliverySupervisor] = {
@@ -102,22 +158,10 @@ class ReliableDeliverySupervisorSpec extends
PekkoSpec(ReliableDeliverySuperviso
system,
new PekkoProtocolSettings(system.settings.config),
PekkoPduProtobufCodec)
- val underlyingHandle =
- TestAssociationHandle(localAddress.copy(protocol = "test"),
remoteAddress.copy(protocol = "test"),
- underlyingTransport, inbound = true)
- val handle = new PekkoProtocolHandle(
- localAddress,
- remoteAddress,
- Promise[AssociationHandle.HandleEventListener](),
- underlyingHandle,
- HandshakeInfo(remoteAddress, initialUid),
- TestProbe().ref,
- PekkoPduProtobufCodec,
- RARP(system).provider.remoteSettings.ProtocolName)
TestActorRef[ReliableDeliverySupervisor](
ReliableDeliverySupervisor.props(
- handleOrActive = Some(handle),
+ handleOrActive = Some(newProtocolHandle(initialUid,
underlyingTransport)),
localAddress,
remoteAddress,
refuseUid = None,
@@ -128,6 +172,35 @@ class ReliableDeliverySupervisorSpec extends
PekkoSpec(ReliableDeliverySuperviso
parent)
}
+ private def expectActive(supervisor: ActorRef): Unit = {
+ val idleProbe = TestProbe()
+ supervisor.tell(ReliableDeliverySupervisor.IsIdle, idleProbe.ref)
+ idleProbe.expectNoMessage(100.millis)
+ }
+
+ private def newProtocolHandle(uid: Int): PekkoProtocolHandle =
+ newProtocolHandle(
+ uid,
+ new TestTransport(
+ localAddress.copy(protocol = "test"),
+ new TestTransport.AssociationRegistry,
+ schemeIdentifier = "test"))
+
+ private def newProtocolHandle(uid: Int, underlyingTransport: TestTransport):
PekkoProtocolHandle = {
+ val underlyingHandle =
+ TestAssociationHandle(localAddress.copy(protocol = "test"),
remoteAddress.copy(protocol = "test"),
+ underlyingTransport, inbound = true)
+ new PekkoProtocolHandle(
+ localAddress,
+ remoteAddress,
+ Promise[AssociationHandle.HandleEventListener](),
+ underlyingHandle,
+ HandshakeInfo(remoteAddress, uid),
+ TestProbe().ref,
+ PekkoPduProtobufCodec,
+ RARP(system).provider.remoteSettings.ProtocolName)
+ }
+
private def bufferWith(seqNumbers: Long*): AckedSendBuffer[Send] =
seqNumbers.foldLeft(new AckedSendBuffer[Send](10)) { (buffer, seq) =>
buffer.buffer(send(seq))
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/classic/RemotingSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/classic/RemotingSpec.scala
index 0fb42195e8..cbed60cd92 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/classic/RemotingSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/classic/RemotingSpec.scala
@@ -688,6 +688,117 @@ class RemotingSpec extends PekkoSpec(RemotingSpec.cfg)
with ImplicitSender with
}
+ "should resume the outbound reader when passive read handoff fails" in {
+ val localAddress = Address("pekko.test", "resume-system1", "localhost",
101)
+ val rawLocalAddress = localAddress.copy(protocol = "test")
+ val remoteAddress = Address("pekko.test", "resume-system2", "localhost",
102)
+ val rawRemoteAddress = remoteAddress.copy(protocol = "test")
+ val remoteUID = 17
+ val receiverName = "handoff-receiver"
+
+ val config = ConfigFactory.parseString(s"""
+ pekko.remote.classic.enabled-transports = ["pekko.remote.classic.test"]
+ pekko.remote.classic.retry-gate-closed-for = 5s
+ pekko.remote.classic.log-remote-lifecycle-events = on
+
+ pekko.remote.classic.test {
+ registry-key = TwHFcESm
+ local-address =
"test://${localAddress.system}@${localAddress.host.get}:${localAddress.port.get}"
+ }
+ """).withFallback(remoteSystem.settings.config)
+ val thisSystem = ActorSystem("this-system", config)
+ muteSystem(thisSystem)
+
+ try {
+ val receiverProbe = TestProbe()(thisSystem)
+ thisSystem.actorOf(Props(new Actor {
+ def receive = {
+ case message => receiverProbe.ref ! message
+ }
+ }).withDeploy(Deploy.local), receiverName)
+ val associationEventProbe = TestProbe()(thisSystem)
+ thisSystem.eventStream.subscribe(associationEventProbe.ref,
classOf[AssociatedEvent])
+
+ val registry = AssociationRegistry.get("TwHFcESm")
+ val remoteTransport = new TestTransport(rawRemoteAddress, registry)
+ val remoteTransportProbe = TestProbe()
+
+ registry.registerTransport(
+ remoteTransport,
+ associationEventListenerFuture = Future.successful(new
Transport.AssociationEventListener {
+ override def notify(ev: Transport.AssociationEvent): Unit =
+ remoteTransportProbe.ref ! ev
+ }))
+
+ awaitCond(registry.transportsReady(rawLocalAddress, rawRemoteAddress))
+
+ val dummySelection =
+
thisSystem.actorSelection(ActorPath.fromString(remoteAddress.toString +
"/user/noonethere"))
+ dummySelection.tell("ping", system.deadLetters)
+
+ val outboundRemoteHandle =
remoteTransportProbe.expectMsgType[Transport.InboundAssociation]
+ val outboundHandleProbe = TestProbe()
+ outboundRemoteHandle.association.readHandlerPromise.success(new
HandleEventListener {
+ override def notify(ev: HandleEvent): Unit = outboundHandleProbe.ref
! ev
+ })
+
+ def handshakePacket(uid: Int): ByteString =
+
PekkoPduProtobufCodec.constructAssociate(HandshakeInfo(rawRemoteAddress, uid))
+ val originalHandshakePacket = handshakePacket(remoteUID)
+ outboundHandleProbe.fishForMessage(3.seconds, "outbound handshake") {
+ case AssociationHandle.InboundPayload(_) => true
+ case _ => false
+ }
+ outboundRemoteHandle.association.write(originalHandshakePacket)
+
+ def payload(message: String): ByteString = {
+ val serializedMessage =
+
MessageSerializer.serialize(thisSystem.asInstanceOf[ExtendedActorSystem],
message)
+ val recipient = new EmptyLocalActorRef(
+ thisSystem.asInstanceOf[ExtendedActorSystem].provider,
+ RootActorPath(localAddress) / "user" / receiverName,
+ thisSystem.eventStream)
+
+ PekkoPduProtobufCodec.constructPayload(
+ PekkoPduProtobufCodec.constructMessage(
+ localAddress,
+ recipient,
+ serializedMessage,
+ pekko.util.OptionVal.None))
+ }
+
+ outboundRemoteHandle.association.write(payload("before-handoff"))
+ receiverProbe.expectMsg("before-handoff")
+
+ val inboundHandleProbe = TestProbe()
+ val inboundHandle =
Await.result(remoteTransport.associate(rawLocalAddress), 3.seconds)
+ inboundHandle.readHandlerPromise.success(new
AssociationHandle.HandleEventListener {
+ override def notify(ev: HandleEvent): Unit = inboundHandleProbe.ref
! ev
+ })
+
+ awaitAssert {
+
registry.getRemoteReadHandlerFor(inboundHandle.asInstanceOf[TestAssociationHandle]).get
+ }
+
+ inboundHandle.write(originalHandshakePacket)
+ associationEventProbe.fishForMessage(3.seconds, "passive association")
{
+ case AssociatedEvent(`localAddress`, `remoteAddress`, true) => true
+ case _ => false
+ }
+
+ val brokenPacket =
PekkoPduProtobufCodec.constructPayload(ByteString(0, 1, 2, 3, 4, 5, 6))
+ inboundHandle.write(brokenPacket)
+ inboundHandleProbe.fishForMessage(3.seconds, "read-only endpoint
disassociation") {
+ case _: AssociationHandle.Disassociated => true
+ case _ => false
+ }
+
+
outboundRemoteHandle.association.write(payload("after-handoff-failure"))
+ receiverProbe.expectMsg("after-handoff-failure")
+ } finally shutdown(thisSystem)
+
+ }
+
"should properly quarantine stashed inbound connections" in {
val localAddress = Address("pekko.test", "system1", "localhost", 1)
val rawLocalAddress = localAddress.copy(protocol = "test")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
