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

use Flow.lazyInit in Artery TCP (#24590)

Browse source
This commit is contained in:
Patrik Nordwall 2018-02-22 11:18:44 +01:00 • committed by Konrad `ktoso` Malawski
parent a89268ee45
commit c15c22622e
1 changed files with 11 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

12
akka-remote/src/main/scala/akka/remote/artery/tcp/ArteryTcpTransport.scala
View file

@ -124,17 +124,19 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
def connectionFlowWithRestart: Flow[ByteString, ByteString, NotUsed] = { def connectionFlowWithRestart: Flow[ByteString, ByteString, NotUsed] = {
val flowFactory = () ⇒ { val flowFactory = () ⇒ {
val flow =
Flow[ByteString]
.via(Flow.lazyInit(_ ⇒ {
// only open the actual connection if any new messages are sent
afr.loFreq( afr.loFreq(
TcpOutbound_Connected, TcpOutbound_Connected,
s"${outboundContext.remoteAddress.host.get}:${outboundContext.remoteAddress.port.get} " + s"${outboundContext.remoteAddress.host.get}:${outboundContext.remoteAddress.port.get} " +
s"/ ${streamName(streamId)}") s"/ ${streamName(streamId)}")
Future.successful(
// FIXME use the Flow.lazyInit from https://github.com/akka/akka/pull/24527
val flow =
Flow[ByteString] Flow[ByteString]
.prepend(Source.single(TcpFraming.encodeConnectionHeader(streamId))) .prepend(Source.single(TcpFraming.encodeConnectionHeader(streamId)))
.via(connectionFlow) .via(connectionFlow))
}, () ⇒ NotUsed))
.recoverWithRetries(1, { case ArteryTransport.ShutdownSignal ⇒ Source.empty }) .recoverWithRetries(1, { case ArteryTransport.ShutdownSignal ⇒ Source.empty })
.log(name = s"outbound connection to [${outboundContext.remoteAddress}], ${streamName(streamId)} stream") .log(name = s"outbound connection to [${outboundContext.remoteAddress}], ${streamName(streamId)} stream")
.addAttributes(Attributes.logLevels(onElement = LogLevels.Off, onFailure = Logging.WarningLevel)) .addAttributes(Attributes.logLevels(onElement = LogLevels.Off, onFailure = Logging.WarningLevel))