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

Decrease visibility of internals to private[io], clean up imports

Browse source
This commit is contained in:
Mathias 2013-01-22 14:10:36 +01:00
parent 79accb810e
commit 570b02f569
9 changed files with 49 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

16
akka-io/src/main/scala/akka/io/Tcp.scala
View file

@ -128,7 +128,7 @@ object Tcp extends ExtensionKey[TcpExt] {
}
/// COMMANDS
sealed trait Command
trait Command
case class Connect(remoteAddress: InetSocketAddress,
localAddress: Option[InetSocketAddress] = None,
@ -180,20 +180,6 @@ object Tcp extends ExtensionKey[TcpExt] {
case object ConfirmedClosed extends ConnectionClosed
case object PeerClosed extends ConnectionClosed
case class ErrorClose(cause: String) extends ConnectionClosed
/// INTERNAL
case class RegisterOutgoingConnection(channel: SocketChannel)
case class RegisterServerSocketChannel(channel: ServerSocketChannel)
case class RegisterIncomingConnection(channel: SocketChannel, handler: ActorRef,
options: Traversable[SocketOption]) extends Command
case class Retry(command: Command, retriesLeft: Int) { require(retriesLeft >= 0) }
case object ChannelConnectable
case object ChannelAcceptable
case object ChannelReadable
case object ChannelWritable
case object AcceptInterest
case object ReadInterest
case object WriteInterest
}
class TcpExt(system: ExtendedActorSystem) extends IO.Extension {

12
akka-io/src/main/scala/akka/io/TcpConnection.scala
View file

@ -7,21 +7,21 @@ package akka.io
import java.net.InetSocketAddress
import java.io.IOException
import java.nio.channels.SocketChannel
import java.nio.ByteBuffer
import scala.annotation.tailrec
import scala.util.control.NonFatal
import scala.collection.immutable
import scala.concurrent.duration._
import akka.actor._
import akka.util.ByteString
import Tcp._
import annotation.tailrec
import java.nio.ByteBuffer
import TcpSelector._
/**
* Base class for TcpIncomingConnection and TcpOutgoingConnection.
*/
abstract class TcpConnection(val selector: ActorRef,
val channel: SocketChannel,
val tcp: TcpExt) extends Actor with ActorLogging with WithBufferPool {
private[io] abstract class TcpConnection(val selector: ActorRef,
val channel: SocketChannel,
val tcp: TcpExt) extends Actor with ActorLogging with WithBufferPool {
import tcp.Settings._
var pendingWrite: PendingWrite = null

11
akka-io/src/main/scala/akka/io/TcpIncomingConnection.scala
View file

@ -5,7 +5,6 @@
package akka.io
import java.nio.channels.SocketChannel
import scala.collection.immutable
import akka.actor.ActorRef
import Tcp.SocketOption
@ -13,11 +12,11 @@ import Tcp.SocketOption
* An actor handling the connection state machine for an incoming, already connected
* SocketChannel.
*/
class TcpIncomingConnection(_selector: ActorRef,
_channel: SocketChannel,
_tcp: TcpExt,
handler: ActorRef,
options: Traversable[SocketOption])
private[io] class TcpIncomingConnection(_selector: ActorRef,
_channel: SocketChannel,
_tcp: TcpExt,
handler: ActorRef,
options: Traversable[SocketOption])
extends TcpConnection(_selector, _channel, _tcp) {
context.watch(handler) // sign death pact

16
akka-io/src/main/scala/akka/io/TcpListener.scala
View file

@ -7,18 +7,18 @@ package akka.io
import java.net.InetSocketAddress
import java.nio.channels.ServerSocketChannel
import scala.annotation.tailrec
import scala.collection.immutable
import scala.util.control.NonFatal
import akka.actor.{ ActorLogging, ActorRef, Actor }
import TcpSelector._
import Tcp._
class TcpListener(selector: ActorRef,
handler: ActorRef,
endpoint: InetSocketAddress,
backlog: Int,
bindCommander: ActorRef,
settings: TcpExt#Settings,
options: Traversable[SocketOption]) extends Actor with ActorLogging {
private[io] class TcpListener(selector: ActorRef,
handler: ActorRef,
endpoint: InetSocketAddress,
backlog: Int,
bindCommander: ActorRef,
settings: TcpExt#Settings,
options: Traversable[SocketOption]) extends Actor with ActorLogging {
context.watch(handler) // sign death pact
val channel = {

2
akka-io/src/main/scala/akka/io/TcpManager.scala
View file

@ -43,7 +43,7 @@ import Tcp._
* with a [[akka.io.Tcp.CommandFailed]] message. This message contains the original command for reference.
*
*/
class TcpManager(tcp: TcpExt) extends Actor with ActorLogging {
private[io] class TcpManager(tcp: TcpExt) extends Actor with ActorLogging {
val selectorPool = context.actorOf(
props = Props(new TcpSelector(self, tcp)).withRouter(RandomRouter(tcp.Settings.NrOfSelectors)),

13
akka-io/src/main/scala/akka/io/TcpOutgoingConnection.scala
View file

@ -8,18 +8,19 @@ import java.net.InetSocketAddress
import java.io.IOException
import java.nio.channels.SocketChannel
import akka.actor.ActorRef
import TcpSelector._
import Tcp._
/**
* An actor handling the connection state machine for an outgoing connection
* to be established.
*/
class TcpOutgoingConnection(_selector: ActorRef,
_tcp: TcpExt,
commander: ActorRef,
remoteAddress: InetSocketAddress,
localAddress: Option[InetSocketAddress],
options: Traversable[SocketOption])
private[io] class TcpOutgoingConnection(_selector: ActorRef,
_tcp: TcpExt,
commander: ActorRef,
remoteAddress: InetSocketAddress,
localAddress: Option[InetSocketAddress],
options: Traversable[SocketOption])
extends TcpConnection(_selector, TcpOutgoingConnection.newSocketChannel(), _tcp) {
context.watch(commander) // sign death pact

19
akka-io/src/main/scala/akka/io/TcpSelector.scala
View file

@ -14,7 +14,8 @@ import scala.concurrent.duration._
import akka.actor._
import Tcp._
class TcpSelector(manager: ActorRef, tcp: TcpExt) extends Actor with ActorLogging {
private[io] class TcpSelector(manager: ActorRef, tcp: TcpExt) extends Actor with ActorLogging {
import TcpSelector._
import tcp.Settings._
@volatile var childrenKeys = HashMap.empty[String, SelectionKey]
@ -223,3 +224,19 @@ class TcpSelector(manager: ActorRef, tcp: TcpExt) extends Actor with ActorLoggin
}
}
}
private[io] object TcpSelector {
case class RegisterOutgoingConnection(channel: SocketChannel)
case class RegisterServerSocketChannel(channel: ServerSocketChannel)
case class RegisterIncomingConnection(channel: SocketChannel, handler: ActorRef,
options: Traversable[SocketOption]) extends Tcp.Command
case class Retry(command: Command, retriesLeft: Int) { require(retriesLeft >= 0) }
case object ChannelConnectable
case object ChannelAcceptable
case object ChannelReadable
case object ChannelWritable
case object AcceptInterest
case object ReadInterest
case object WriteInterest
}

1
akka-io/src/test/scala/akka/io/TcpConnectionSpec.scala
View file

@ -18,6 +18,7 @@ import akka.actor.{ PoisonPill, ActorRef, Terminated }
import akka.testkit.{ TestProbe, TestActorRef, AkkaSpec }
import akka.util.ByteString
import TestUtils._
import TcpSelector._
import Tcp._
class TcpConnectionSpec extends AkkaSpec("akka.io.tcp.register-timeout = 500ms") {

1
akka-io/src/test/scala/akka/io/TcpListenerSpec.scala
View file

@ -8,6 +8,7 @@ import java.net.Socket
import scala.concurrent.duration._
import akka.actor.{ Terminated, SupervisorStrategy, Actor, Props }
import akka.testkit.{ TestProbe, TestActorRef, AkkaSpec }
import TcpSelector._
import Tcp._
class TcpListenerSpec extends AkkaSpec("akka.io.tcp.batch-accept-limit = 2") {