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

Add Artery TCP transport, #24390

Browse source 
* transport config
* TCP specific classes in akka.remote.artery.tcp package
* TcpFraming stage that handle the additional streamId field and length based framing.
  Credit to jrudolph for this clean solution, which made it possible to use same
  envolope header for Aeron and TCP.
* magic first bytes to detect invalid access
* drain SendQueue to deadLetters in postStop
* error handling, restart, inbound and outbound streams
* udp vs tcp in autoSelectPort
* TCP specific flight recorder events
* update reference documentation
This commit is contained in:
Patrik Nordwall 2018-02-16 09:26:29 +01:00
parent 0d222906f4
commit 162a1f80a0
13 changed files with 790 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

15
akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
View file

@ -9,11 +9,12 @@ import akka.actor._
import akka.dispatch.sysmsg._
import akka.event.{ EventStream, Logging, LoggingAdapter }
import akka.event.Logging.Error
import akka.serialization.{ Serialization, SerializationExtension }
import akka.pattern.pipe
import scala.util.control.NonFatal
import akka.actor.SystemGuardian.{ RegisterTerminationHook, TerminationHook, TerminationHookDone }
import scala.util.control.Exception.Catcher
import scala.concurrent.Future
@ -21,12 +22,15 @@ import akka.ConfigurationException
import akka.annotation.InternalApi
import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics }
import akka.remote.artery.ArteryTransport
import akka.remote.artery.aeron.ArteryAeronUdpTransport
import akka.remote.artery.ArterySettings
import akka.remote.artery.ArterySettings.AeronUpd
import akka.util.OptionVal
import akka.remote.artery.OutboundEnvelope
import akka.remote.artery.SystemMessageDelivery.SystemMessageEnvelope
import akka.remote.artery.aeron.ArteryAeronUdpTransport
import akka.remote.serialization.ActorRefResolveCache
import akka.remote.serialization.ActorRefResolveThreadLocalCache
import akka.remote.artery.tcp.ArteryTcpTransport
/**
* INTERNAL API
@ -204,7 +208,12 @@ private[akka] class RemoteActorRefProvider(
local.registerExtraNames(Map(("remote", d)))
d
},
transport = if (remoteSettings.Artery.Enabled) new ArteryAeronUdpTransport(system, this) else new Remoting(system, this))
transport =
if (remoteSettings.Artery.Enabled) remoteSettings.Artery.Transport match {
case ArterySettings.AeronUpd ⇒ new ArteryAeronUdpTransport(system, this)
case ArterySettings.Tcp ⇒ new ArteryTcpTransport(system, this)
}
else new Remoting(system, this))
_internals = internals
remotingTerminator ! internals