I am handling connections with the following code:
val connections = Tcp().bind(interface, port)
connections runForeach { conn ⇒
log.info(s"Client connected from: ${conn.remoteAddress}")
try {
val flow = MTProto.flow(nextConnId(), maxBufferSize, sessionRegion)
conn.handleWith(flow)
} catch {
case e: Exception ⇒
log.error(e, "Failed to create connection flow")
}
}
where flow is:
def flow(connId: String, sessionRegion: SessionRegion)(implicit db: Database,
system: ActorSystem, timeout: Timeout) = {
val authManager = system.actorOf(AuthorizationManager.props(db),
s"authManager-${connId}")
val authSource = Source(ActorPublisher[MTProto](authManager))
val sessionClient = system.actorOf(SessionClient.props(sessionRegion),
s"sessionClient-${connId}")
val sessionClientSource = Source(ActorPublisher[MTProto](sessionClient))
val mtprotoFlow = Flow[ByteString]
.transform(() ⇒ new PackageParseStage)
.transform(() ⇒ new PackageCheckStage)
.transform(() ⇒ new PackageHandleStage(protoVersions, apiMajorVersions,
authManager, sessionClient))
val mapRespFlow: Flow[MTProto, ByteString, Unit] = Flow[MTProto]
.transform(() ⇒ mapResponse(system))
val completeSink = Sink.onComplete {
case x ⇒
system.log.debug("Completing {}", x)
}
Flow() { implicit builder ⇒
import FlowGraph.Implicits._
val bcast = builder.add(Broadcast[ByteString](2))
val merge = builder.add(Merge[MTProto](3))
val mtproto = builder.add(mtprotoFlow)
val auth = builder.add(authSource)
val session = builder.add(sessionClientSource)
val mapResp = builder.add(mapRespFlow)
val complete = builder.add(completeSink)
// format: OFF
mtproto ~> merge
auth ~> merge
session ~> merge ~> mapResp ~> bcast ~> complete
// format: ON
(mtproto.inlet, bcast.out(1))
}
}
- PackageParseStage is a StatefulStage which parses ByteStrings into
protocol entities
- PackageCheckStage is a PushStage which makes some checks on protocol
entities and makes ctx.fail if somethig is wrong
- PackageHandleStage is a StatefulStage which forwards requests to some
external services and generates some packages to send to client via
emit(iterator.ctx)
- auth and sessions are ActorPublishers which publish responses from
external services for sending to a client.
On Sunday, June 7, 2015 at 7:03:01 AM UTC+3, Andrey Kuznetsov wrote:
>
> I am using akka-streams for handling incoming TCP connections. After
> putting application behind AWS Elastic Load Balancer, it started to fall
> into "too many open files" problem because of lots of connections in
> CLOSE_WAIT state. Is there an ability to close such connection on a server
> side?
>
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
