diff --git a/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala index ae6532acea..66714125c9 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala @@ -318,17 +318,20 @@ class ActorRefSpec extends AkkaSpec with DefaultTimeout { val out = new ObjectOutputStream(baos) val sysImpl = system.asInstanceOf[ActorSystemImpl] - val addr = sysImpl.provider.rootPath.address - val serialized = SerializedActorRef(RootActorPath(addr, "/non-existing")) + val ref = system.actorOf(Props[ReplyActor], "non-existing") + val serialized = SerializedActorRef(ref) out.writeObject(serialized) out.flush out.close + ref ! PoisonPill + awaitCond(ref.isTerminated, 2 seconds) + JavaSerializer.currentSystem.withValue(sysImpl) { val in = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray)) - in.readObject must be === new EmptyLocalActorRef(sysImpl.provider, system.actorFor("/").path / "non-existing", system.eventStream) + in.readObject must be === new EmptyLocalActorRef(sysImpl.provider, ref.path, system.eventStream) } } @@ -403,7 +406,7 @@ class ActorRefSpec extends AkkaSpec with DefaultTimeout { Await.result(ffive, timeout.duration) must be("five") Await.result(fnull, timeout.duration) must be("null") - awaitCond(ref.isTerminated, 2000 millis) + awaitCond(ref.isTerminated, 2 seconds) } "restart when Kill:ed" in { diff --git a/akka-actor/src/main/scala/akka/actor/ActorRef.scala b/akka-actor/src/main/scala/akka/actor/ActorRef.scala index fe0f66f5f5..99eae0fff4 100644 --- a/akka-actor/src/main/scala/akka/actor/ActorRef.scala +++ b/akka-actor/src/main/scala/akka/actor/ActorRef.scala @@ -6,14 +6,12 @@ package akka.actor import akka.dispatch._ import akka.dispatch.sysmsg._ -import akka.util._ import java.lang.{ UnsupportedOperationException, IllegalStateException } import akka.serialization.{ Serialization, JavaSerializer } import akka.event.EventStream import scala.annotation.tailrec import java.util.concurrent.ConcurrentHashMap import akka.event.LoggingAdapter -import scala.collection.JavaConverters /** * Immutable and serializable handle to an actor, which may or may not reside @@ -381,7 +379,7 @@ private[akka] class LocalActorRef private[akka] ( override def restart(cause: Throwable): Unit = actorCell.restart(cause) @throws(classOf[java.io.ObjectStreamException]) - protected def writeReplace(): AnyRef = SerializedActorRef(path) + protected def writeReplace(): AnyRef = SerializedActorRef(this) } /** @@ -407,11 +405,8 @@ private[akka] case class SerializedActorRef private (path: String) { * INTERNAL API */ private[akka] object SerializedActorRef { - def apply(path: ActorPath): SerializedActorRef = { - Serialization.currentTransportAddress.value match { - case null ⇒ new SerializedActorRef(path.toSerializationFormat) - case addr ⇒ new SerializedActorRef(path.toSerializationFormatWithAddress(addr)) - } + def apply(actorRef: ActorRef): SerializedActorRef = { + new SerializedActorRef(Serialization.serializedActorPath(actorRef)) } } @@ -437,7 +432,7 @@ private[akka] trait MinimalActorRef extends InternalActorRef with LocalRef { override def restart(cause: Throwable): Unit = () @throws(classOf[java.io.ObjectStreamException]) - protected def writeReplace(): AnyRef = SerializedActorRef(path) + protected def writeReplace(): AnyRef = SerializedActorRef(this) } /** diff --git a/akka-actor/src/main/scala/akka/actor/RepointableActorRef.scala b/akka-actor/src/main/scala/akka/actor/RepointableActorRef.scala index 123a576462..9ce7169be3 100644 --- a/akka-actor/src/main/scala/akka/actor/RepointableActorRef.scala +++ b/akka-actor/src/main/scala/akka/actor/RepointableActorRef.scala @@ -157,7 +157,7 @@ private[akka] class RepointableActorRef( def sendSystemMessage(message: SystemMessage) = underlying.sendSystemMessage(message) @throws(classOf[java.io.ObjectStreamException]) - protected def writeReplace(): AnyRef = SerializedActorRef(path) + protected def writeReplace(): AnyRef = SerializedActorRef(this) } private[akka] class UnstartedCell(val systemImpl: ActorSystemImpl, diff --git a/akka-actor/src/main/scala/akka/serialization/Serialization.scala b/akka-actor/src/main/scala/akka/serialization/Serialization.scala index ef21355502..fbe4933dcf 100644 --- a/akka-actor/src/main/scala/akka/serialization/Serialization.scala +++ b/akka-actor/src/main/scala/akka/serialization/Serialization.scala @@ -5,7 +5,7 @@ package akka.serialization import com.typesafe.config.Config -import akka.actor.{ Extension, ExtendedActorSystem, Address } +import akka.actor._ import akka.event.Logging import java.util.concurrent.ConcurrentHashMap import scala.collection.mutable.ArrayBuffer @@ -21,10 +21,10 @@ object Serialization { type ClassSerializer = (Class[_], Serializer) /** - * This holds a reference to the current transport address to be inserted - * into local actor refs during serialization. + * This holds a reference to the current transport serialization information used for + * serializing local actor refs. */ - val currentTransportAddress = new DynamicVariable[Address](null) + val currentTransportInformation = new DynamicVariable[SerializationInformation](null) class Settings(val config: Config) { val Serializers: Map[String, String] = configToMap("akka.actor.serializers") @@ -35,8 +35,34 @@ object Serialization { config.getConfig(path).root.unwrapped.asScala.toMap map { case (k, v) ⇒ (k -> v.toString) } } } + + /** + * The serialized path of an actorRef, based on the current transport serialization information. + */ + def serializedActorPath(actorRef: ActorRef): String = { + val path = actorRef.path + val originalSystem: ActorSystemImpl = actorRef match { + case a: ActorRefWithCell ⇒ a.underlying.systemImpl + case _ ⇒ null + } + Serialization.currentTransportInformation.value match { + case null ⇒ path.toSerializationFormat + case SerializationInformation(address, system) ⇒ + if (originalSystem == null || originalSystem == system) + path.toSerializationFormatWithAddress(address) + else { + val provider = originalSystem.provider + path.toSerializationFormatWithAddress(provider.getExternalAddressFor(address).getOrElse(provider.getDefaultAddress)) + } + } + } } +/** + * Serialization information needed for serializing local actor refs. + */ +case class SerializationInformation(val address: Address, val system: ActorSystem) + /** * Serialization module. Contains methods for serialization and deserialization as well as * locating a Serializer for a particular class as defined in the mapping in the configuration. diff --git a/akka-docs/rst/java/code/docs/serialization/SerializationDocTestBase.java b/akka-docs/rst/java/code/docs/serialization/SerializationDocTestBase.java index 3eac0502d2..7c8a1bda69 100644 --- a/akka-docs/rst/java/code/docs/serialization/SerializationDocTestBase.java +++ b/akka-docs/rst/java/code/docs/serialization/SerializationDocTestBase.java @@ -7,7 +7,6 @@ import org.junit.Test; import static org.junit.Assert.*; //#imports import akka.actor.*; -import akka.remote.RemoteActorRefProvider; import akka.serialization.*; //#imports @@ -58,20 +57,19 @@ public class SerializationDocTestBase { //#actorref-serializer // Serialize // (beneath toBinary) - final Address transportAddress = - Serialization.currentTransportAddress().value(); + final SerializationInformation info = Serialization.currentTransportInformation().value(); + String identifier; - // If there is no transportAddress, - // it means that either this Serializer isn't called + // If there is no SerializationInformation, + // it means that this Serializer isn't called // within a piece of code that sets it, // so either you need to supply your own, // or simply use the local path. - if (transportAddress == null) identifier = theActorRef.path().toSerializationFormat(); - else identifier = theActorRef.path().toSerializationFormatWithAddress(transportAddress); + if (info == null) identifier = theActorRef.path().toSerializationFormat(); + else identifier = Serialization.serializedActorPath(theActorRef); // Then just serialize the identifier however you like - // Deserialize // (beneath fromBinary) final ActorRef deserializedActorRef = theActorSystem.actorFor(identifier); diff --git a/akka-docs/rst/scala/code/docs/serialization/SerializationDocSpec.scala b/akka-docs/rst/scala/code/docs/serialization/SerializationDocSpec.scala index 0dacdeff3b..7b0a4fce28 100644 --- a/akka-docs/rst/scala/code/docs/serialization/SerializationDocSpec.scala +++ b/akka-docs/rst/scala/code/docs/serialization/SerializationDocSpec.scala @@ -165,14 +165,14 @@ package docs.serialization { // Serialize // (beneath toBinary) - // If there is no transportAddress, - // it means that either this Serializer isn't called + // If there is no SerializationInformation, + // it means that this Serializer isn't called // within a piece of code that sets it, - // so either you need to supply your own, + // so either you need to supply your own address, // or simply use the local path. - val identifier: String = Serialization.currentTransportAddress.value match { - case null ⇒ theActorRef.path.toSerializationFormat - case address ⇒ theActorRef.path.toSerializationFormatWithAddress(address) + val identifier: String = Serialization.currentTransportInformation.value match { + case null ⇒ theActorRef.path.toSerializationFormat + case _: SerializationInformation ⇒ Serialization.serializedActorPath(theActorRef) } // Then just serialize the identifier however you like diff --git a/akka-remote/src/main/scala/akka/remote/Endpoint.scala b/akka-remote/src/main/scala/akka/remote/Endpoint.scala index 9b870e0f46..2a7a8368cf 100644 --- a/akka-remote/src/main/scala/akka/remote/Endpoint.scala +++ b/akka-remote/src/main/scala/akka/remote/Endpoint.scala @@ -13,7 +13,7 @@ import akka.remote.RemoteProtocol.MessageProtocol import akka.remote.transport.AkkaPduCodec._ import akka.remote.transport.AssociationHandle._ import akka.remote.transport.{ AkkaPduCodec, Transport, AssociationHandle } -import akka.serialization.Serialization +import akka.serialization.{ SerializationInformation, Serialization } import akka.util.ByteString import akka.remote.transport.Transport.InvalidAssociationException import java.io.NotSerializableException @@ -332,7 +332,7 @@ private[remote] class EndpointWriter( private def serializeMessage(msg: Any): MessageProtocol = handle match { case Some(h) ⇒ - Serialization.currentTransportAddress.withValue(h.localAddress) { + Serialization.currentTransportInformation.withValue(SerializationInformation(h.localAddress, context.system)) { (MessageSerializer.serialize(extendedSystem, msg.asInstanceOf[AnyRef])) } case None ⇒ throw new EndpointException("Internal error: No handle was present during serialization of" + diff --git a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala index 4ac2b95229..867e59d859 100644 --- a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala +++ b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala @@ -271,8 +271,15 @@ private[akka] class RemoteActorRefProvider( def actorFor(ref: InternalActorRef, path: String): InternalActorRef = path match { case ActorPathExtractor(address, elems) ⇒ if (hasAddress(address)) actorFor(rootGuardian, elems) - else new RemoteActorRef(transport, transport.localAddressForRemote(address), - new RootActorPath(address) / elems, Nobody, props = None, deploy = None) + else try { + new RemoteActorRef(transport, transport.localAddressForRemote(address), + new RootActorPath(address) / elems, Nobody, props = None, deploy = None) + } catch { + case NonFatal(e) ⇒ + val oldPath = RootActorPath(address) / elems + log.error(e, "Error while looking up address {}", oldPath.address) + new EmptyLocalActorRef(this, oldPath, eventStream) + } case _ ⇒ local.actorFor(ref, path) } @@ -378,5 +385,5 @@ private[akka] class RemoteActorRef private[akka] ( def restart(cause: Throwable): Unit = sendSystemMessage(Recreate(cause)) @throws(classOf[java.io.ObjectStreamException]) - private def writeReplace(): AnyRef = SerializedActorRef(path) + private def writeReplace(): AnyRef = SerializedActorRef(this) } diff --git a/akka-remote/src/main/scala/akka/remote/RemoteTransport.scala b/akka-remote/src/main/scala/akka/remote/RemoteTransport.scala index c697068b07..6725383c49 100644 --- a/akka-remote/src/main/scala/akka/remote/RemoteTransport.scala +++ b/akka-remote/src/main/scala/akka/remote/RemoteTransport.scala @@ -22,7 +22,7 @@ class RemoteTransportException(message: String, cause: Throwable) extends AkkaEx * * The remote transport is responsible for sending and receiving messages. * Each transport has an address, which it should provide in - * Serialization.currentTransportAddress (thread-local) while serializing + * Serialization.currentTransportInformation (thread-local) while serializing * actor references (which might also be part of messages). This address must * be available (i.e. fully initialized) by the time the first message is * received or when the start() method returns, whatever happens first. diff --git a/akka-remote/src/main/scala/akka/remote/Remoting.scala b/akka-remote/src/main/scala/akka/remote/Remoting.scala index 1fe6fc21a7..9be6deb426 100644 --- a/akka-remote/src/main/scala/akka/remote/Remoting.scala +++ b/akka-remote/src/main/scala/akka/remote/Remoting.scala @@ -74,7 +74,7 @@ private[remote] object Remoting { null) } case None ⇒ throw new RemoteTransportException( - s"No transport is loaded for protocol: [${remote.protocol}], available protocols: [${transportMapping.keys.mkString}]", null) + s"No transport is loaded for protocol: [${remote.protocol}], available protocols: [${transportMapping.keys.mkString(", ")}]", null) } } diff --git a/akka-remote/src/main/scala/akka/remote/serialization/ProtobufSerializer.scala b/akka-remote/src/main/scala/akka/remote/serialization/ProtobufSerializer.scala index 056439c23e..acbbc26cf2 100644 --- a/akka-remote/src/main/scala/akka/remote/serialization/ProtobufSerializer.scala +++ b/akka-remote/src/main/scala/akka/remote/serialization/ProtobufSerializer.scala @@ -6,10 +6,8 @@ package akka.remote.serialization import akka.serialization.{ Serializer, Serialization } import com.google.protobuf.Message -import akka.actor.DynamicAccess +import akka.actor.{ ActorSystem, ActorRef } import akka.remote.RemoteProtocol.ActorRefProtocol -import akka.actor.ActorSystem -import akka.actor.ActorRef object ProtobufSerializer { @@ -18,11 +16,7 @@ object ProtobufSerializer { * protobuf representation. */ def serializeActorRef(ref: ActorRef): ActorRefProtocol = { - val identifier: String = Serialization.currentTransportAddress.value match { - case null ⇒ ref.path.toSerializationFormat - case address ⇒ ref.path.toSerializationFormatWithAddress(address) - } - ActorRefProtocol.newBuilder.setPath(identifier).build + ActorRefProtocol.newBuilder.setPath(Serialization.serializedActorPath(ref)).build } /** diff --git a/akka-remote/src/test/scala/akka/remote/RemotingSpec.scala b/akka-remote/src/test/scala/akka/remote/RemotingSpec.scala index 37be9b5553..6fafb50d98 100644 --- a/akka-remote/src/test/scala/akka/remote/RemotingSpec.scala +++ b/akka-remote/src/test/scala/akka/remote/RemotingSpec.scala @@ -37,14 +37,24 @@ object RemotingSpec { class Echo2 extends Actor { def receive = { - case "ping" ⇒ sender ! (("pong", sender)) + case "ping" ⇒ sender ! (("pong", sender)) + case a: ActorRef ⇒ a ! (("ping", sender)) + case ("ping", a: ActorRef) ⇒ sender ! (("pong", a)) + case ("pong", a: ActorRef) ⇒ a ! (("pong", sender.path.toSerializationFormat)) } } - val cfg: Config = ConfigFactory parseString (""" + class Proxy(val one: ActorRef, val another: ActorRef) extends Actor { + def receive = { + case s if sender.path == one.path ⇒ another ! s + case s if sender.path == another.path ⇒ one ! s + } + } + + val cfg: Config = ConfigFactory parseString (s""" common-ssl-settings { - key-store = "%s" - trust-store = "%s" + key-store = "${getClass.getClassLoader.getResource("keystore").getPath}" + trust-store = "${getClass.getClassLoader.getResource("truststore").getPath}" key-store-password = "changeme" trust-store-password = "changeme" protocol = "TLSv1" @@ -83,10 +93,10 @@ object RemotingSpec { } } - netty.tcp = ${common-netty-settings} - netty.udp = ${common-netty-settings} - netty.ssl = ${common-netty-settings} - netty.ssl.security = ${common-ssl-settings} + netty.tcp = $${common-netty-settings} + netty.udp = $${common-netty-settings} + netty.ssl = $${common-netty-settings} + netty.ssl.security = $${common-ssl-settings} test { transport-class = "akka.remote.transport.TestTransport" @@ -104,9 +114,7 @@ object RemotingSpec { /looker/child/grandchild.remote = "akka.test://RemotingSpec@localhost:12345" } } - """.format( - getClass.getClassLoader.getResource("keystore").getPath, - getClass.getClassLoader.getResource("truststore").getPath)) + """) } @@ -122,14 +130,14 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D maximum-payload-bytes = 48000 bytes } """).withFallback(system.settings.config).resolve() - val otherSystem = ActorSystem("remote-sys", conf) + val remoteSystem = ActorSystem("remote-sys", conf) for ( (name, proto) ← Seq( "/gonk" -> "tcp", "/zagzag" -> "udp", "/roghtaar" -> "ssl.tcp") - ) deploy(system, Deploy(name, scope = RemoteScope(addr(otherSystem, proto)))) + ) deploy(system, Deploy(name, scope = RemoteScope(addr(remoteSystem, proto)))) def addr(sys: ActorSystem, proto: String) = sys.asInstanceOf[ExtendedActorSystem].provider.getExternalAddressFor(Address(s"akka.$proto", "", "", 0)).get @@ -138,12 +146,12 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D sys.asInstanceOf[ExtendedActorSystem].provider.asInstanceOf[RemoteActorRefProvider].deployer.deploy(d) } - val remote = otherSystem.actorOf(Props[Echo2], "echo") + val remote = remoteSystem.actorOf(Props[Echo2], "echo") val here = system.actorFor("akka.test://remote-sys@localhost:12346/user/echo") private def verifySend(msg: Any)(afterSend: ⇒ Unit) { - val bigBounceOther = otherSystem.actorOf(Props(new Actor { + val bigBounceOther = remoteSystem.actorOf(Props(new Actor { def receive = { case x: Int ⇒ sender ! byteStringOfSize(x) case x ⇒ sender ! x @@ -166,16 +174,26 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D system.eventStream.unsubscribe(eventForwarder, classOf[AssociationErrorEvent]) system.eventStream.unsubscribe(eventForwarder, classOf[DisassociatedEvent]) system.stop(eventForwarder) - otherSystem.stop(bigBounceOther) + remoteSystem.stop(bigBounceOther) } } + override def atStartup() = { + system.eventStream.publish(TestEvent.Mute( + EventFilter.error(start = "AssociationError"), + EventFilter.warning(pattern = "received dead letter.*"))) + remoteSystem.eventStream.publish(TestEvent.Mute( + EventFilter[EndpointException](), + EventFilter.error(start = "AssociationError"), + EventFilter.warning(pattern = "received dead letter.*(InboundPayload|Disassociate|HandleListener)"))) + } + private def byteStringOfSize(size: Int) = ByteString.fromArray(Array.fill(size)(42: Byte)) val maxPayloadBytes = system.settings.config.getBytes("akka.remote.test.maximum-payload-bytes").toInt override def afterTermination() { - otherSystem.shutdown() + remoteSystem.shutdown() AssociationRegistry.clear() } @@ -203,16 +221,21 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D "send dead letters on remote if actor does not exist" in { EventFilter.warning(pattern = "dead.*buh", occurrences = 1).intercept { system.actorFor("akka.test://remote-sys@localhost:12346/does/not/exist") ! "buh" - }(otherSystem) + }(remoteSystem) } "not be exhausted by sending to broken connections" in { val tcpOnlyConfig = ConfigFactory.parseString("""akka.remote.enabled-transports = ["akka.remote.netty.tcp"]"""). - withFallback(otherSystem.settings.config) - val moreSystems = Vector.fill(5)(ActorSystem(otherSystem.name, tcpOnlyConfig)) - moreSystems foreach (_.actorOf(Props[Echo2], name = "echo")) + withFallback(remoteSystem.settings.config) + val moreSystems = Vector.fill(5)(ActorSystem(remoteSystem.name, tcpOnlyConfig)) + moreSystems foreach { sys ⇒ + sys.eventStream.publish(TestEvent.Mute( + EventFilter[EndpointDisassociatedException](), + EventFilter.warning(pattern = "received dead letter.*"))) + sys.actorOf(Props[Echo2], name = "echo") + } val moreRefs = moreSystems map (sys ⇒ system.actorFor(RootActorPath(addr(sys, "tcp")) / "user" / "echo")) - val aliveEcho = system.actorFor(RootActorPath(addr(otherSystem, "tcp")) / "user" / "echo") + val aliveEcho = system.actorFor(RootActorPath(addr(remoteSystem, "tcp")) / "user" / "echo") val n = 100 // first everything is up and running @@ -259,7 +282,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D } "not send to remote re-created actor with same name" in { - val echo = otherSystem.actorOf(Props[Echo1], "otherEcho1") + val echo = remoteSystem.actorOf(Props[Echo1], "otherEcho1") echo ! 71 expectMsg(71) echo ! PoisonPill @@ -267,7 +290,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D echo ! 72 expectNoMsg(1.second) - val echo2 = otherSystem.actorOf(Props[Echo1], "otherEcho1") + val echo2 = remoteSystem.actorOf(Props[Echo1], "otherEcho1") echo2 ! 73 expectMsg(73) // msg to old ActorRef (different uid) should not get through @@ -275,7 +298,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D echo ! 74 expectNoMsg(1.second) - otherSystem.actorFor("/user/otherEcho1") ! 75 + remoteSystem.actorFor("/user/otherEcho1") ! 75 expectMsg(75) system.actorFor("akka.test://remote-sys@localhost:12346/user/otherEcho1") ! 76 @@ -289,11 +312,11 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D case s: String ⇒ sender ! context.actorFor(s) } }), "looker") - // child is configured to be deployed on remote-sys (otherSystem) - l ! (Props[Echo1], "child") + // child is configured to be deployed on remote-sys (remoteSystem) + l ! ((Props[Echo1], "child")) val child = expectMsgType[ActorRef] // grandchild is configured to be deployed on RemotingSpec (system) - child ! (Props[Echo1], "grandchild") + child ! ((Props[Echo1], "grandchild")) val grandchild = expectMsgType[ActorRef] grandchild.asInstanceOf[ActorRefScope].isLocal must be(true) grandchild ! 43 @@ -313,7 +336,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D child ! PoisonPill expectMsg("postStop") expectMsgType[Terminated].actor must be === child - l ! (Props[Echo1], "child") + l ! ((Props[Echo1], "child")) val child2 = expectMsgType[ActorRef] child2 ! 45 expectMsg(45) @@ -335,7 +358,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D "be able to use multiple transports and use the appropriate one (TCP)" in { val r = system.actorOf(Props[Echo1], "gonk") r.path.toString must be === - s"akka.tcp://remote-sys@localhost:${port(otherSystem, "tcp")}/remote/akka.tcp/RemotingSpec@localhost:${port(system, "tcp")}/user/gonk" + s"akka.tcp://remote-sys@localhost:${port(remoteSystem, "tcp")}/remote/akka.tcp/RemotingSpec@localhost:${port(system, "tcp")}/user/gonk" r ! 42 expectMsg(42) EventFilter[Exception]("crash", occurrences = 1).intercept { @@ -351,7 +374,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D "be able to use multiple transports and use the appropriate one (UDP)" in { val r = system.actorOf(Props[Echo1], "zagzag") r.path.toString must be === - s"akka.udp://remote-sys@localhost:${port(otherSystem, "udp")}/remote/akka.udp/RemotingSpec@localhost:${port(system, "udp")}/user/zagzag" + s"akka.udp://remote-sys@localhost:${port(remoteSystem, "udp")}/remote/akka.udp/RemotingSpec@localhost:${port(system, "udp")}/user/zagzag" r ! 42 expectMsg(10.seconds, 42) EventFilter[Exception]("crash", occurrences = 1).intercept { @@ -367,7 +390,7 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D "be able to use multiple transports and use the appropriate one (SSL)" in { val r = system.actorOf(Props[Echo1], "roghtaar") r.path.toString must be === - s"akka.ssl.tcp://remote-sys@localhost:${port(otherSystem, "ssl.tcp")}/remote/akka.ssl.tcp/RemotingSpec@localhost:${port(system, "ssl.tcp")}/user/roghtaar" + s"akka.ssl.tcp://remote-sys@localhost:${port(remoteSystem, "ssl.tcp")}/remote/akka.ssl.tcp/RemotingSpec@localhost:${port(system, "ssl.tcp")}/user/roghtaar" r ! 42 expectMsg(10.seconds, 42) EventFilter[Exception]("crash", occurrences = 1).intercept { @@ -415,15 +438,31 @@ class RemotingSpec extends AkkaSpec(RemotingSpec.cfg) with ImplicitSender with D } } + "be able to serialize a local actor ref from another actor system" in { + val config = ConfigFactory.parseString(""" + akka.remote.enabled-transports = ["akka.remote.test", "akka.remote.netty.tcp"] + akka.remote.test.local-address = "test://other-system@localhost:12347" + """).withFallback(remoteSystem.settings.config) + val otherSystem = ActorSystem("other-system", config) + try { + val otherGuy = otherSystem.actorOf(Props[Echo2], "other-guy") + // check that we use the specified transport address instead of the default + val otherGuyRemoteTcp = otherGuy.path.toSerializationFormatWithAddress(addr(otherSystem, "tcp")) + val remoteEchoHereTcp = system.actorFor(s"akka.tcp://remote-sys@localhost:${port(remoteSystem, "tcp")}/user/echo") + val proxyTcp = system.actorOf(Props(new Proxy(remoteEchoHereTcp, self)), "proxy-tcp") + proxyTcp ! otherGuy + expectMsg(3.seconds, ("pong", otherGuyRemoteTcp)) + // now check that we fall back to default when we haven't got a corresponding transport + val otherGuyRemoteTest = otherGuy.path.toSerializationFormatWithAddress(addr(otherSystem, "test")) + val remoteEchoHereSsl = system.actorFor(s"akka.ssl.tcp://remote-sys@localhost:${port(remoteSystem, "ssl.tcp")}/user/echo") + val proxySsl = system.actorOf(Props(new Proxy(remoteEchoHereSsl, self)), "proxy-ssl") + proxySsl ! otherGuy + expectMsg(3.seconds, ("pong", otherGuyRemoteTest)) + } finally { + otherSystem.shutdown() + otherSystem.awaitTermination(5.seconds.dilated) + otherSystem.isTerminated must be(true) + } + } } - - override def beforeTermination() { - system.eventStream.publish(TestEvent.Mute( - EventFilter.warning(pattern = "received dead letter.*(InboundPayload|Disassociate)"))) - otherSystem.eventStream.publish(TestEvent.Mute( - EventFilter[EndpointException](), - EventFilter.error(start = "AssociationError"), - EventFilter.warning(pattern = "received dead letter.*(InboundPayload|Disassociate|HandleListener)"))) - } - }