!act #3670: Removing old IO implementation
This commit is contained in:
Endre Sándor Varga
parent
3fdf20dfc6
commit
b543644baa
8 changed files with 1 additions and 2278 deletions
|
|
@ -1,418 +0,0 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
package akka.actor
|
||||
|
||||
import language.postfixOps
|
||||
import akka.util.ByteString
|
||||
import scala.concurrent.{ ExecutionContext, Await, Future, Promise }
|
||||
import scala.concurrent.duration._
|
||||
import scala.util.continuations._
|
||||
import akka.testkit._
|
||||
import akka.dispatch.MessageDispatcher
|
||||
import akka.pattern.ask
|
||||
import java.net.{ Socket, InetSocketAddress, InetAddress, SocketAddress }
|
||||
import scala.util.Failure
|
||||
import scala.annotation.tailrec
|
||||
import akka.AkkaException
|
||||
|
||||
object IOActorSpec {
|
||||
|
||||
class SimpleEchoServer(addressPromise: Promise[SocketAddress]) extends Actor {
|
||||
|
||||
val server = IOManager(context.system) listen ("localhost", 0)
|
||||
|
||||
val state = IO.IterateeRef.Map.sync[IO.Handle]()
|
||||
|
||||
def receive = {
|
||||
|
||||
case IO.Listening(`server`, address) ⇒
|
||||
addressPromise success address
|
||||
|
||||
case IO.NewClient(`server`) ⇒
|
||||
val socket = server.accept()
|
||||
state(socket) flatMap (_ ⇒ IO repeat (IO.takeAny map socket.write))
|
||||
|
||||
case IO.Read(socket, bytes) ⇒
|
||||
state(socket)(IO Chunk bytes)
|
||||
|
||||
case IO.Closed(socket, cause) ⇒
|
||||
state -= socket
|
||||
|
||||
}
|
||||
|
||||
override def postStop {
|
||||
server.close()
|
||||
state.keySet foreach (_.close())
|
||||
}
|
||||
}
|
||||
|
||||
class SimpleEchoClient(address: SocketAddress) extends Actor {
|
||||
|
||||
val socket = IOManager(context.system) connect (address)
|
||||
|
||||
val state = IO.IterateeRef.sync()
|
||||
|
||||
def receive = {
|
||||
|
||||
case _: IO.Connected ⇒ //don't care
|
||||
|
||||
case bytes: ByteString ⇒
|
||||
val source = sender
|
||||
socket write bytes
|
||||
state flatMap { _ ⇒ IO take bytes.length map (source ! _) }
|
||||
|
||||
case IO.Read(`socket`, bytes) ⇒
|
||||
state(IO Chunk bytes)
|
||||
|
||||
case IO.Closed(`socket`, cause) ⇒
|
||||
state(cause)
|
||||
cause match {
|
||||
case IO.Error(e) ⇒ throw e
|
||||
case _ ⇒ throw new RuntimeException("Socket closed")
|
||||
}
|
||||
}
|
||||
|
||||
override def postStop {
|
||||
socket.close()
|
||||
state(IO EOF)
|
||||
}
|
||||
}
|
||||
|
||||
sealed trait KVCommand {
|
||||
def bytes: ByteString
|
||||
}
|
||||
|
||||
case class KVSet(key: String, value: String) extends KVCommand {
|
||||
val bytes = ByteString("SET " + key + " " + value.length + "\r\n" + value + "\r\n")
|
||||
}
|
||||
|
||||
case class KVGet(key: String) extends KVCommand {
|
||||
val bytes = ByteString("GET " + key + "\r\n")
|
||||
}
|
||||
|
||||
case object KVGetAll extends KVCommand {
|
||||
val bytes = ByteString("GETALL\r\n")
|
||||
}
|
||||
|
||||
// Basic Redis-style protocol
|
||||
class KVStore(addressPromise: Promise[SocketAddress]) extends Actor {
|
||||
|
||||
import context.system
|
||||
|
||||
val state = IO.IterateeRef.Map.sync[IO.Handle]()
|
||||
|
||||
var kvs: Map[String, String] = Map.empty
|
||||
|
||||
val server = IOManager(context.system) listen ("localhost", 0)
|
||||
|
||||
val EOL = ByteString("\r\n")
|
||||
|
||||
def receive = {
|
||||
|
||||
case IO.Listening(`server`, address) ⇒
|
||||
addressPromise success address
|
||||
|
||||
case IO.NewClient(`server`) ⇒
|
||||
val socket = server.accept()
|
||||
state(socket) flatMap { _ ⇒
|
||||
IO repeat {
|
||||
IO takeUntil EOL map (_.utf8String split ' ') flatMap {
|
||||
|
||||
case Array("SET", key, length) ⇒
|
||||
for {
|
||||
value ← IO take length.toInt
|
||||
_ ← IO takeUntil EOL
|
||||
} yield {
|
||||
kvs += (key -> value.utf8String)
|
||||
ByteString("+OK\r\n")
|
||||
}
|
||||
|
||||
case Array("GET", key) ⇒
|
||||
IO Iteratee {
|
||||
kvs get key map { value ⇒
|
||||
ByteString("$" + value.length + "\r\n" + value + "\r\n")
|
||||
} getOrElse ByteString("$-1\r\n")
|
||||
}
|
||||
|
||||
case Array("GETALL") ⇒
|
||||
IO Iteratee {
|
||||
(ByteString("*" + (kvs.size * 2) + "\r\n") /: kvs) {
|
||||
case (result, (k, v)) ⇒
|
||||
val kBytes = ByteString(k)
|
||||
val vBytes = ByteString(v)
|
||||
result ++
|
||||
ByteString("$" + kBytes.length) ++ EOL ++
|
||||
kBytes ++ EOL ++
|
||||
ByteString("$" + vBytes.length) ++ EOL ++
|
||||
vBytes ++ EOL
|
||||
}
|
||||
}
|
||||
|
||||
} map (socket write)
|
||||
}
|
||||
}
|
||||
|
||||
case IO.Read(socket, bytes) ⇒
|
||||
state(socket)(IO Chunk bytes)
|
||||
|
||||
case _: IO.Connected ⇒ //don't care
|
||||
|
||||
case IO.Closed(socket, cause) ⇒
|
||||
state -= socket
|
||||
|
||||
}
|
||||
|
||||
override def postStop {
|
||||
server.close()
|
||||
state.keySet foreach (_.close())
|
||||
}
|
||||
}
|
||||
|
||||
class KVClient(address: SocketAddress) extends Actor {
|
||||
|
||||
val socket = IOManager(context.system) connect (address)
|
||||
|
||||
val state = IO.IterateeRef.sync()
|
||||
|
||||
val EOL = ByteString("\r\n")
|
||||
|
||||
def receive = {
|
||||
case cmd: KVCommand ⇒
|
||||
val source = sender
|
||||
socket write cmd.bytes
|
||||
state flatMap { _ ⇒
|
||||
readResult map (source !)
|
||||
}
|
||||
|
||||
case _: IO.Connected ⇒ //don't care
|
||||
|
||||
case IO.Read(`socket`, bytes) ⇒
|
||||
state(IO Chunk bytes)
|
||||
|
||||
case IO.Closed(`socket`, cause) ⇒
|
||||
state(cause)
|
||||
throw cause match {
|
||||
case IO.Error(t) ⇒ t
|
||||
case _ ⇒ new RuntimeException("Socket closed")
|
||||
}
|
||||
}
|
||||
|
||||
override def postStop {
|
||||
socket.close()
|
||||
state(IO.EOF)
|
||||
}
|
||||
|
||||
def readResult: IO.Iteratee[Any] = {
|
||||
IO take 1 map (_.utf8String) flatMap {
|
||||
case "+" ⇒ IO takeUntil EOL map (msg ⇒ msg.utf8String)
|
||||
case "-" ⇒ IO takeUntil EOL flatMap (err ⇒ IO.Failure(new RuntimeException(err.utf8String)))
|
||||
case "$" ⇒
|
||||
IO takeUntil EOL map (_.utf8String.toInt) flatMap {
|
||||
case -1 ⇒ IO Done None
|
||||
case length ⇒
|
||||
for {
|
||||
value ← IO take length
|
||||
_ ← IO takeUntil EOL
|
||||
} yield Some(value.utf8String)
|
||||
}
|
||||
case "*" ⇒
|
||||
IO takeUntil EOL map (_.utf8String.toInt) flatMap {
|
||||
case -1 ⇒ IO Done None
|
||||
case length ⇒
|
||||
IO.takeList(length)(readResult) flatMap { list ⇒
|
||||
((Right(Map()): Either[String, Map[String, String]]) /: list.grouped(2)) {
|
||||
case (Right(m), List(Some(k: String), Some(v: String))) ⇒ Right(m + (k -> v))
|
||||
case (Right(_), _) ⇒ Left("Unexpected Response")
|
||||
case (left, _) ⇒ left
|
||||
} fold (msg ⇒ IO.Failure(new RuntimeException(msg)), IO Done _)
|
||||
}
|
||||
}
|
||||
case _ ⇒ IO.Failure(new RuntimeException("Unexpected Response"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
|
||||
class IOActorSpec extends AkkaSpec with DefaultTimeout {
|
||||
import IOActorSpec._
|
||||
|
||||
/**
|
||||
* Retries the future until a result is returned or until one of the limits are hit. If no
|
||||
* limits are provided the future will be retried indefinitely until a result is returned.
|
||||
*
|
||||
* @param count number of retries
|
||||
* @param timeout duration to retry within
|
||||
* @param delay duration to wait before retrying
|
||||
* @param filter determines which exceptions should be retried
|
||||
* @return a future containing the result or the last exception before a limit was hit.
|
||||
*/
|
||||
def retry[T](count: Option[Int] = None,
|
||||
timeout: Option[FiniteDuration] = None,
|
||||
delay: Option[FiniteDuration] = Some(100 millis),
|
||||
filter: Option[Throwable ⇒ Boolean] = None)(future: ⇒ Future[T])(implicit executor: ExecutionContext): Future[T] = {
|
||||
|
||||
val promise = Promise[T]()
|
||||
|
||||
val timer: Option[Deadline] = timeout match {
|
||||
case Some(duration) ⇒ Some(duration fromNow)
|
||||
case None ⇒ None
|
||||
}
|
||||
|
||||
def check(n: Int, e: Throwable): Boolean =
|
||||
(count.isEmpty || (n < count.get)) && (timer.isEmpty || timer.get.hasTimeLeft()) && (filter.isEmpty || filter.get(e))
|
||||
|
||||
def run(n: Int) {
|
||||
future onComplete {
|
||||
case Failure(e) if check(n, e) ⇒
|
||||
if (delay.isDefined) {
|
||||
executor match {
|
||||
case m: MessageDispatcher ⇒ m.configurator.prerequisites.scheduler.scheduleOnce(delay.get)(run(n + 1))
|
||||
case _ ⇒ // Thread.sleep, ignore, or other?
|
||||
}
|
||||
} else run(n + 1)
|
||||
case v ⇒ promise complete v
|
||||
}
|
||||
}
|
||||
|
||||
run(0)
|
||||
|
||||
promise.future
|
||||
}
|
||||
|
||||
"an IO Actor" must {
|
||||
import system.dispatcher
|
||||
"run echo server" in {
|
||||
filterException[java.net.ConnectException] {
|
||||
val addressPromise = Promise[SocketAddress]()
|
||||
val server = system.actorOf(Props(new SimpleEchoServer(addressPromise)))
|
||||
val address = Await.result(addressPromise.future, TestLatch.DefaultTimeout)
|
||||
val client = system.actorOf(Props(new SimpleEchoClient(address)))
|
||||
val f1 = retry() { client ? ByteString("Hello World!1") }
|
||||
val f2 = retry() { client ? ByteString("Hello World!2") }
|
||||
val f3 = retry() { client ? ByteString("Hello World!3") }
|
||||
Await.result(f1, TestLatch.DefaultTimeout) must equal(ByteString("Hello World!1"))
|
||||
Await.result(f2, TestLatch.DefaultTimeout) must equal(ByteString("Hello World!2"))
|
||||
Await.result(f3, TestLatch.DefaultTimeout) must equal(ByteString("Hello World!3"))
|
||||
system.stop(client)
|
||||
system.stop(server)
|
||||
}
|
||||
}
|
||||
|
||||
"run echo server under high load" in {
|
||||
filterException[java.net.ConnectException] {
|
||||
val addressPromise = Promise[SocketAddress]()
|
||||
val server = system.actorOf(Props(new SimpleEchoServer(addressPromise)))
|
||||
val address = Await.result(addressPromise.future, TestLatch.DefaultTimeout)
|
||||
val client = system.actorOf(Props(new SimpleEchoClient(address)))
|
||||
val list = List.range(0, 100)
|
||||
val f = Future.traverse(list)(i ⇒ retry() { client ? ByteString(i.toString) })
|
||||
assert(Await.result(f, TestLatch.DefaultTimeout).size === 100)
|
||||
system.stop(client)
|
||||
system.stop(server)
|
||||
}
|
||||
}
|
||||
|
||||
"run key-value store" in {
|
||||
filterException[java.net.ConnectException] {
|
||||
val addressPromise = Promise[SocketAddress]()
|
||||
val server = system.actorOf(Props(new KVStore(addressPromise)))
|
||||
val address = Await.result(addressPromise.future, TestLatch.DefaultTimeout)
|
||||
val client1 = system.actorOf(Props(new KVClient(address)))
|
||||
val client2 = system.actorOf(Props(new KVClient(address)))
|
||||
val f1 = retry() { client1 ? KVSet("hello", "World") }
|
||||
val f2 = retry() { client1 ? KVSet("test", "No one will read me") }
|
||||
val f3 = f1 flatMap { _ ⇒ retry() { client1 ? KVGet("hello") } }
|
||||
val f4 = f2 flatMap { _ ⇒ retry() { client2 ? KVSet("test", "I'm a test!") } }
|
||||
val f5 = f4 flatMap { _ ⇒ retry() { client1 ? KVGet("test") } }
|
||||
val f6 = Future.sequence(List(f3, f5)) flatMap { _ ⇒ retry() { client2 ? KVGetAll } }
|
||||
Await.result(f1, TestLatch.DefaultTimeout) must equal("OK")
|
||||
Await.result(f2, TestLatch.DefaultTimeout) must equal("OK")
|
||||
Await.result(f3, TestLatch.DefaultTimeout) must equal(Some("World"))
|
||||
Await.result(f4, TestLatch.DefaultTimeout) must equal("OK")
|
||||
Await.result(f5, TestLatch.DefaultTimeout) must equal(Some("I'm a test!"))
|
||||
Await.result(f6, TestLatch.DefaultTimeout) must equal(Map("hello" -> "World", "test" -> "I'm a test!"))
|
||||
system.stop(client1)
|
||||
system.stop(client2)
|
||||
system.stop(server)
|
||||
}
|
||||
}
|
||||
|
||||
"takeUntil must fail on EOF before predicate when used with repeat" in {
|
||||
val CRLF = ByteString("\r\n")
|
||||
val dest = new InetSocketAddress(InetAddress.getLocalHost.getHostAddress, { val s = new java.net.ServerSocket(0); try s.getLocalPort finally s.close() })
|
||||
|
||||
val a = system.actorOf(Props(new Actor {
|
||||
val state = IO.IterateeRef.Map.async[IO.Handle]()(context.dispatcher)
|
||||
|
||||
override def preStart {
|
||||
IOManager(context.system) listen dest
|
||||
}
|
||||
|
||||
def receive = {
|
||||
case _: IO.Listening ⇒ testActor ! "Wejkipejki"
|
||||
case IO.NewClient(server) ⇒
|
||||
val socket = server.accept()
|
||||
state(socket) flatMap (_ ⇒ IO.repeat(for (input ← IO.takeUntil(CRLF)) yield testActor ! input.utf8String))
|
||||
|
||||
case IO.Read(socket, bytes) ⇒
|
||||
state(socket)(IO Chunk bytes)
|
||||
|
||||
case IO.Closed(socket, cause) ⇒
|
||||
state(socket)(IO EOF)
|
||||
state -= socket
|
||||
testActor ! "eof"
|
||||
}
|
||||
}))
|
||||
expectMsg("Wejkipejki")
|
||||
val s = new Socket(dest.getAddress, dest.getPort)
|
||||
try {
|
||||
val expectedReceive = Seq("ole", "dole", "doff", "kinke", "lane", "koff", "ole", "dole", "dinke", "dane", "ole", "dole")
|
||||
val expectedSend = expectedReceive ++ Seq("doff")
|
||||
val out = s.getOutputStream
|
||||
out.write(expectedSend.mkString(CRLF.utf8String).getBytes("UTF-8"))
|
||||
out.flush()
|
||||
for (word ← expectedReceive) expectMsg(word)
|
||||
s.close()
|
||||
expectMsg("eof")
|
||||
} finally {
|
||||
if (!s.isClosed) s.close()
|
||||
}
|
||||
}
|
||||
|
||||
"fail when listening on an invalid address" in {
|
||||
implicit val self = testActor
|
||||
val address = new InetSocketAddress("irate.elephant", 9999)
|
||||
IOManager(system).listen(address)
|
||||
expectMsgType[Status.Failure](1 seconds)
|
||||
}
|
||||
|
||||
"fail when listening on a privileged port" in {
|
||||
implicit val self = testActor
|
||||
val address = new InetSocketAddress("localhost", 80) // Assumes test not run as root
|
||||
IOManager(system).listen(address)
|
||||
expectMsgType[Status.Failure](1 seconds)
|
||||
}
|
||||
|
||||
"fail when connecting to an invalid address" in {
|
||||
implicit val self = testActor
|
||||
val address = new InetSocketAddress("irate.elephant", 80)
|
||||
IOManager(system).connect(address)
|
||||
expectMsgType[Status.Failure](1 seconds)
|
||||
}
|
||||
|
||||
"fail when binding to already bound port and report port in failure" in {
|
||||
implicit val self = testActor
|
||||
IOManager(system).listen(new InetSocketAddress("localhost", 0))
|
||||
val boundTo = expectMsgType[IO.Listening].address.asInstanceOf[InetSocketAddress]
|
||||
IOManager(system).listen(boundTo)
|
||||
val exc = expectMsgType[Status.Failure].cause
|
||||
exc.getClass must be(classOf[AkkaException])
|
||||
exc.getMessage must include(boundTo.getPort.toString)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -9,7 +9,7 @@ import akka.testkit.AkkaSpec
|
|||
import com.typesafe.config.ConfigFactory
|
||||
import scala.collection.JavaConverters._
|
||||
import scala.concurrent.duration._
|
||||
import akka.actor.{ IOManager, ActorSystem }
|
||||
import akka.actor.ActorSystem
|
||||
import akka.event.Logging.DefaultLogger
|
||||
|
||||
@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
|
||||
|
|
@ -126,19 +126,6 @@ class ConfigSpec extends AkkaSpec(ConfigFactory.defaultReference(ActorSystem.fin
|
|||
settings.DebugRouterMisconfiguration must be(false)
|
||||
}
|
||||
|
||||
// IO config
|
||||
{
|
||||
val io = config.getConfig("akka.io")
|
||||
val ioExtSettings = IOManager(system).settings
|
||||
ioExtSettings.readBufferSize must be(8192)
|
||||
io.getBytes("read-buffer-size") must be(ioExtSettings.readBufferSize)
|
||||
|
||||
ioExtSettings.selectInterval must be(100)
|
||||
io.getInt("select-interval") must be(ioExtSettings.selectInterval)
|
||||
|
||||
ioExtSettings.defaultBacklog must be(1000)
|
||||
io.getInt("default-backlog") must be(ioExtSettings.defaultBacklog)
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
|
|
|
|||
|
|
@ -702,24 +702,6 @@ akka {
|
|||
management-dispatcher = "akka.actor.default-dispatcher"
|
||||
}
|
||||
|
||||
|
||||
# IMPORTANT NOTICE:
|
||||
#
|
||||
# The following settings belong to the deprecated akka.actor.IO
|
||||
# implementation and will be removed once that is removed. They are not
|
||||
# taken into account by the akka.io.* implementation, which is configured
|
||||
# above!
|
||||
|
||||
# In bytes, the size of the shared read buffer. In the span 0b..2GiB.
|
||||
#
|
||||
read-buffer-size = 8KiB
|
||||
|
||||
# Specifies how many ops are done between every descriptor selection
|
||||
select-interval = 100
|
||||
|
||||
# Number of connections that are allowed in the backlog.
|
||||
# 0 or negative means that the platform default will be used.
|
||||
default-backlog = 1000
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -1,84 +0,0 @@
|
|||
/**
|
||||
* Copyright (C) 2013 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package docs.io
|
||||
|
||||
//#imports
|
||||
import akka.actor._
|
||||
import akka.util.{ ByteString, ByteStringBuilder, ByteIterator }
|
||||
//#imports
|
||||
|
||||
abstract class BinaryDecoding {
|
||||
//#decoding
|
||||
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
|
||||
|
||||
val FrameDecoder = for {
|
||||
frameLenBytes ← IO.take(4)
|
||||
frameLen = frameLenBytes.iterator.getInt
|
||||
frame ← IO.take(frameLen)
|
||||
} yield {
|
||||
val in = frame.iterator
|
||||
|
||||
val n = in.getInt
|
||||
val m = in.getInt
|
||||
|
||||
val a = Array.newBuilder[Short]
|
||||
val b = Array.newBuilder[Long]
|
||||
|
||||
for (i ← 1 to n) {
|
||||
a += in.getShort
|
||||
b += in.getInt
|
||||
}
|
||||
|
||||
val data = Array.ofDim[Double](m)
|
||||
in.getDoubles(data)
|
||||
|
||||
(a.result, b.result, data)
|
||||
}
|
||||
|
||||
//#decoding
|
||||
}
|
||||
|
||||
abstract class RestToSeq {
|
||||
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
|
||||
val bytes: ByteString
|
||||
val in = bytes.iterator
|
||||
|
||||
//#rest-to-seq
|
||||
val n = in.getInt
|
||||
val m = in.getInt
|
||||
// ... in.get...
|
||||
val rest: ByteString = in.toSeq
|
||||
//#rest-to-seq
|
||||
}
|
||||
|
||||
abstract class BinaryEncoding {
|
||||
//#encoding
|
||||
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
|
||||
|
||||
val a: Array[Short]
|
||||
val b: Array[Long]
|
||||
val data: Array[Double]
|
||||
|
||||
val frameBuilder = ByteString.newBuilder
|
||||
|
||||
val n = a.length
|
||||
val m = data.length
|
||||
|
||||
frameBuilder.putInt(n)
|
||||
frameBuilder.putInt(m)
|
||||
|
||||
for (i ← 0 to n - 1) {
|
||||
frameBuilder.putShort(a(i))
|
||||
frameBuilder.putLong(b(i))
|
||||
}
|
||||
frameBuilder.putDoubles(data)
|
||||
val frame = frameBuilder.result()
|
||||
//#encoding
|
||||
|
||||
//#sending
|
||||
val socket: IO.SocketHandle
|
||||
socket.write(ByteString.newBuilder.putInt(frame.length).result)
|
||||
socket.write(frame)
|
||||
//#sending
|
||||
}
|
||||
|
|
@ -1,242 +0,0 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package docs.io
|
||||
|
||||
import language.postfixOps
|
||||
|
||||
//#imports
|
||||
import akka.actor._
|
||||
import akka.util.{ ByteString, ByteStringBuilder }
|
||||
import java.net.InetSocketAddress
|
||||
//#imports
|
||||
|
||||
//#actor
|
||||
class HttpServer(port: Int) extends Actor {
|
||||
|
||||
val state = IO.IterateeRef.Map.async[IO.Handle]()(context.dispatcher)
|
||||
|
||||
override def preStart {
|
||||
IOManager(context.system) listen new InetSocketAddress(port)
|
||||
}
|
||||
|
||||
def receive = {
|
||||
|
||||
case IO.NewClient(server) ⇒
|
||||
val socket = server.accept()
|
||||
state(socket) flatMap (_ ⇒ HttpServer.processRequest(socket))
|
||||
|
||||
case IO.Read(socket, bytes) ⇒
|
||||
state(socket)(IO Chunk bytes)
|
||||
|
||||
case IO.Closed(socket, cause) ⇒
|
||||
state(socket)(IO EOF)
|
||||
state -= socket
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
//#actor
|
||||
|
||||
//#actor-companion
|
||||
object HttpServer {
|
||||
import HttpIteratees._
|
||||
|
||||
def processRequest(socket: IO.SocketHandle): IO.Iteratee[Unit] =
|
||||
IO repeat {
|
||||
for {
|
||||
request ← readRequest
|
||||
} yield {
|
||||
val rsp = request match {
|
||||
case Request("GET", "ping" :: Nil, _, _, headers, _) ⇒
|
||||
OKResponse(ByteString("<p>pong</p>"),
|
||||
request.headers.exists {
|
||||
case Header(n, v) ⇒
|
||||
n.toLowerCase == "connection" && v.toLowerCase == "keep-alive"
|
||||
})
|
||||
case req ⇒
|
||||
OKResponse(ByteString("<p>" + req.toString + "</p>"),
|
||||
request.headers.exists {
|
||||
case Header(n, v) ⇒
|
||||
n.toLowerCase == "connection" && v.toLowerCase == "keep-alive"
|
||||
})
|
||||
}
|
||||
socket write OKResponse.bytes(rsp).compact
|
||||
if (!rsp.keepAlive) socket.close()
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
//#actor-companion
|
||||
|
||||
//#request-class
|
||||
case class Request(meth: String, path: List[String], query: Option[String],
|
||||
httpver: String, headers: List[Header], body: Option[ByteString])
|
||||
case class Header(name: String, value: String)
|
||||
//#request-class
|
||||
|
||||
//#constants
|
||||
object HttpConstants {
|
||||
val SP = ByteString(" ")
|
||||
val HT = ByteString("\t")
|
||||
val CRLF = ByteString("\r\n")
|
||||
val COLON = ByteString(":")
|
||||
val PERCENT = ByteString("%")
|
||||
val PATH = ByteString("/")
|
||||
val QUERY = ByteString("?")
|
||||
}
|
||||
//#constants
|
||||
|
||||
//#read-request
|
||||
object HttpIteratees {
|
||||
import HttpConstants._
|
||||
|
||||
def readRequest =
|
||||
for {
|
||||
requestLine ← readRequestLine
|
||||
(meth, (path, query), httpver) = requestLine
|
||||
headers ← readHeaders
|
||||
body ← readBody(headers)
|
||||
} yield Request(meth, path, query, httpver, headers, body)
|
||||
//#read-request
|
||||
|
||||
//#read-request-line
|
||||
def ascii(bytes: ByteString): String = bytes.decodeString("US-ASCII").trim
|
||||
|
||||
def readRequestLine =
|
||||
for {
|
||||
meth ← IO takeUntil SP
|
||||
uri ← readRequestURI
|
||||
_ ← IO takeUntil SP // ignore the rest
|
||||
httpver ← IO takeUntil CRLF
|
||||
} yield (ascii(meth), uri, ascii(httpver))
|
||||
//#read-request-line
|
||||
|
||||
//#read-request-uri
|
||||
def readRequestURI = IO peek 1 flatMap {
|
||||
case PATH ⇒
|
||||
for {
|
||||
path ← readPath
|
||||
query ← readQuery
|
||||
} yield (path, query)
|
||||
case _ ⇒ sys.error("Not Implemented")
|
||||
}
|
||||
//#read-request-uri
|
||||
|
||||
//#read-path
|
||||
def readPath = {
|
||||
def step(segments: List[String]): IO.Iteratee[List[String]] =
|
||||
IO peek 1 flatMap {
|
||||
case PATH ⇒ IO drop 1 flatMap (_ ⇒ readUriPart(pathchar) flatMap (
|
||||
segment ⇒ step(segment :: segments)))
|
||||
case _ ⇒ segments match {
|
||||
case "" :: rest ⇒ IO Done rest.reverse
|
||||
case _ ⇒ IO Done segments.reverse
|
||||
}
|
||||
}
|
||||
step(Nil)
|
||||
}
|
||||
//#read-path
|
||||
|
||||
//#read-query
|
||||
def readQuery: IO.Iteratee[Option[String]] = IO peek 1 flatMap {
|
||||
case QUERY ⇒ IO drop 1 flatMap (_ ⇒ readUriPart(querychar) map (Some(_)))
|
||||
case _ ⇒ IO Done None
|
||||
}
|
||||
//#read-query
|
||||
|
||||
//#read-uri-part
|
||||
val alpha = Set.empty ++ ('a' to 'z') ++ ('A' to 'Z') map (_.toByte)
|
||||
val digit = Set.empty ++ ('0' to '9') map (_.toByte)
|
||||
val hexdigit = digit ++ (Set.empty ++ ('a' to 'f') ++ ('A' to 'F') map (_.toByte))
|
||||
val subdelim = Set('!', '$', '&', '\'', '(', ')', '*', '+', ',', ';', '=') map
|
||||
(_.toByte)
|
||||
val pathchar = alpha ++ digit ++ subdelim ++ (Set(':', '@') map (_.toByte))
|
||||
val querychar = pathchar ++ (Set('/', '?') map (_.toByte))
|
||||
|
||||
def readUriPart(allowed: Set[Byte]): IO.Iteratee[String] = for {
|
||||
str ← IO takeWhile allowed map ascii
|
||||
pchar ← IO peek 1 map (_ == PERCENT)
|
||||
all ← if (pchar) readPChar flatMap (ch ⇒ readUriPart(allowed) map
|
||||
(str + ch + _))
|
||||
else IO Done str
|
||||
} yield all
|
||||
|
||||
def readPChar = IO take 3 map {
|
||||
case Seq('%', rest @ _*) if rest forall hexdigit ⇒
|
||||
java.lang.Integer.parseInt(rest map (_.toChar) mkString, 16).toChar
|
||||
}
|
||||
//#read-uri-part
|
||||
|
||||
//#read-headers
|
||||
def readHeaders = {
|
||||
def step(found: List[Header]): IO.Iteratee[List[Header]] = {
|
||||
IO peek 2 flatMap {
|
||||
case CRLF ⇒ IO takeUntil CRLF flatMap (_ ⇒ IO Done found)
|
||||
case _ ⇒ readHeader flatMap (header ⇒ step(header :: found))
|
||||
}
|
||||
}
|
||||
step(Nil)
|
||||
}
|
||||
|
||||
def readHeader =
|
||||
for {
|
||||
name ← IO takeUntil COLON
|
||||
value ← IO takeUntil CRLF flatMap readMultiLineValue
|
||||
} yield Header(ascii(name), ascii(value))
|
||||
|
||||
def readMultiLineValue(initial: ByteString): IO.Iteratee[ByteString] =
|
||||
IO peek 1 flatMap {
|
||||
case SP ⇒ IO takeUntil CRLF flatMap (
|
||||
bytes ⇒ readMultiLineValue(initial ++ bytes))
|
||||
case _ ⇒ IO Done initial
|
||||
}
|
||||
//#read-headers
|
||||
|
||||
//#read-body
|
||||
def readBody(headers: List[Header]) =
|
||||
if (headers.exists(header ⇒ header.name == "Content-Length" ||
|
||||
header.name == "Transfer-Encoding"))
|
||||
IO.takeAll map (Some(_))
|
||||
else
|
||||
IO Done None
|
||||
//#read-body
|
||||
}
|
||||
|
||||
//#ok-response
|
||||
object OKResponse {
|
||||
import HttpConstants.CRLF
|
||||
|
||||
val okStatus = ByteString("HTTP/1.1 200 OK")
|
||||
val contentType = ByteString("Content-Type: text/html; charset=utf-8")
|
||||
val cacheControl = ByteString("Cache-Control: no-cache")
|
||||
val date = ByteString("Date: ")
|
||||
val server = ByteString("Server: Akka")
|
||||
val contentLength = ByteString("Content-Length: ")
|
||||
val connection = ByteString("Connection: ")
|
||||
val keepAlive = ByteString("Keep-Alive")
|
||||
val close = ByteString("Close")
|
||||
|
||||
def bytes(rsp: OKResponse) = {
|
||||
new ByteStringBuilder ++=
|
||||
okStatus ++= CRLF ++=
|
||||
contentType ++= CRLF ++=
|
||||
cacheControl ++= CRLF ++=
|
||||
date ++= ByteString(new java.util.Date().toString) ++= CRLF ++=
|
||||
server ++= CRLF ++=
|
||||
contentLength ++= ByteString(rsp.body.length.toString) ++= CRLF ++=
|
||||
connection ++= (if (rsp.keepAlive) keepAlive else close) ++= CRLF ++=
|
||||
CRLF ++= rsp.body result
|
||||
}
|
||||
|
||||
}
|
||||
case class OKResponse(body: ByteString, keepAlive: Boolean)
|
||||
//#ok-response
|
||||
|
||||
//#main
|
||||
object Main extends App {
|
||||
val port = Option(System.getenv("PORT")) map (_.toInt) getOrElse 8080
|
||||
val system = ActorSystem()
|
||||
val server = system.actorOf(Props(classOf[HttpServer], port))
|
||||
}
|
||||
//#main
|
||||
|
|
@ -1,247 +0,0 @@
|
|||
.. _io-scala-old:
|
||||
|
||||
Old IO
|
||||
==============
|
||||
|
||||
.. warning::
|
||||
This is the documentation of the old IO implementation that is considered now deprecated. Please take a look
|
||||
at new IO API: :ref:`io-scala`
|
||||
|
||||
Introduction
|
||||
------------
|
||||
|
||||
This documentation is in progress and some sections may be incomplete. More will be coming.
|
||||
|
||||
Components
|
||||
----------
|
||||
|
||||
ByteString
|
||||
^^^^^^^^^^
|
||||
|
||||
A primary goal of Akka's IO support is to only communicate between actors with immutable objects. When dealing with network IO on the jvm ``Array[Byte]`` and ``ByteBuffer`` are commonly used to represent collections of ``Byte``\s, but they are mutable. Scala's collection library also lacks a suitably efficient immutable collection for ``Byte``\s. Being able to safely and efficiently move ``Byte``\s around is very important for this IO support, so ``ByteString`` was developed.
|
||||
|
||||
``ByteString`` is a `Rope-like <http://en.wikipedia.org/wiki/Rope_(computer_science)>`_ data structure that is immutable and efficient. When 2 ``ByteString``\s are concatenated together they are both stored within the resulting ``ByteString`` instead of copying both to a new ``Array``. Operations such as ``drop`` and ``take`` return ``ByteString``\s that still reference the original ``Array``, but just change the offset and length that is visible. Great care has also been taken to make sure that the internal ``Array`` cannot be modified. Whenever a potentially unsafe ``Array`` is used to create a new ``ByteString`` a defensive copy is created. If you require a ``ByteString`` that only blocks a much memory as necessary for it's content, use the ``compact`` method to get a ``CompactByteString`` instance. If the ``ByteString`` represented only a slice of the original array, this will result in copying all bytes in that slice.
|
||||
|
||||
``ByteString`` inherits all methods from ``IndexedSeq``, and it also has some new ones. For more information, look up the ``akka.util.ByteString`` class and it's companion object in the ScalaDoc.
|
||||
|
||||
``ByteString`` also comes with it's own optimized builder and iterator classes ``ByteStringBuilder`` and ``ByteIterator`` which provides special features in addition to the standard builder / iterator methods:
|
||||
|
||||
Compatibility with java.io
|
||||
..........................
|
||||
|
||||
A ``ByteStringBuilder`` can be wrapped in a `java.io.OutputStream` via the ``asOutputStream`` method. Likewise, ``ByteIterator`` can we wrapped in a ``java.io.InputStream`` via ``asInputStream``. Using these, ``akka.io`` applications can integrate legacy code based on ``java.io`` streams.
|
||||
|
||||
Encoding and decoding of binary data
|
||||
....................................
|
||||
|
||||
``ByteStringBuilder`` and ``ByteIterator`` support encoding and decoding of binary data. As an example, consider a stream of binary data frames with the following format:
|
||||
|
||||
.. code-block:: text
|
||||
|
||||
frameLen: Int
|
||||
n: Int
|
||||
m: Int
|
||||
n times {
|
||||
a: Short
|
||||
b: Long
|
||||
}
|
||||
data: m times Double
|
||||
|
||||
In this example, the data is to be stored in arrays of ``a``, ``b`` and ``data``.
|
||||
|
||||
Decoding of such frames can be efficiently implemented in the following fashion:
|
||||
|
||||
.. includecode:: code/docs/io/BinaryCoding.scala
|
||||
:include: decoding
|
||||
|
||||
This implementation naturally follows the example data format. In a true Scala application, one might, of course, want use specialized immutable Short/Long/Double containers instead of mutable Arrays.
|
||||
|
||||
After extracting data from a ``ByteIterator``, the remaining content can also be turned back into a ``ByteString`` using the ``toSeq`` method
|
||||
|
||||
.. includecode:: code/docs/io/BinaryCoding.scala
|
||||
:include: rest-to-seq
|
||||
|
||||
with no copying from bytes to rest involved. In general, conversions from ByteString to ByteIterator and vice versa are O(1) for non-chunked ByteStrings and (at worst) O(nChunks) for chunked ByteStrings.
|
||||
|
||||
Encoding of data also is very natural, using ``ByteStringBuilder``
|
||||
|
||||
.. includecode:: code/docs/io/BinaryCoding.scala
|
||||
:include: encoding
|
||||
|
||||
|
||||
The encoded data then can be sent over socket (see ``IOManager``):
|
||||
|
||||
.. includecode:: code/docs/io/BinaryCoding.scala
|
||||
:include: sending
|
||||
|
||||
|
||||
IO.Handle
|
||||
^^^^^^^^^
|
||||
|
||||
``IO.Handle`` is an immutable reference to a Java NIO ``Channel``. Passing mutable ``Channel``\s between ``Actor``\s could lead to unsafe behavior, so instead subclasses of the ``IO.Handle`` trait are used. Currently there are 2 concrete subclasses: ``IO.SocketHandle`` (representing a ``SocketChannel``) and ``IO.ServerHandle`` (representing a ``ServerSocketChannel``).
|
||||
|
||||
IOManager
|
||||
^^^^^^^^^
|
||||
|
||||
The ``IOManager`` takes care of the low level IO details. Each ``ActorSystem`` has it's own ``IOManager``, which can be accessed calling ``IOManager(system: ActorSystem)``. ``Actor``\s communicate with the ``IOManager`` with specific messages. The messages sent from an ``Actor`` to the ``IOManager`` are handled automatically when using certain methods and the messages sent from an ``IOManager`` are handled within an ``Actor``\'s ``receive`` method.
|
||||
|
||||
Connecting to a remote host:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
val address = new InetSocketAddress("remotehost", 80)
|
||||
val socket = IOManager(actorSystem).connect(address)
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
val socket = IOManager(actorSystem).connect("remotehost", 80)
|
||||
|
||||
Creating a server:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
val address = new InetSocketAddress("localhost", 80)
|
||||
val serverSocket = IOManager(actorSystem).listen(address)
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
val serverSocket = IOManager(actorSystem).listen("localhost", 80)
|
||||
|
||||
Receiving messages from the ``IOManager``:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
def receive = {
|
||||
|
||||
case IO.Listening(server, address) =>
|
||||
println("The server is listening on socket " + address)
|
||||
|
||||
case IO.Connected(socket, address) =>
|
||||
println("Successfully connected to " + address)
|
||||
|
||||
case IO.NewClient(server) =>
|
||||
println("New incoming connection on server")
|
||||
val socket = server.accept()
|
||||
println("Writing to new client socket")
|
||||
socket.write(bytes)
|
||||
println("Closing socket")
|
||||
socket.close()
|
||||
|
||||
case IO.Read(socket, bytes) =>
|
||||
println("Received incoming data from socket")
|
||||
|
||||
case IO.Closed(socket: IO.SocketHandle, cause) =>
|
||||
println("Socket has closed, cause: " + cause)
|
||||
|
||||
case IO.Closed(server: IO.ServerHandle, cause) =>
|
||||
println("Server socket has closed, cause: " + cause)
|
||||
|
||||
}
|
||||
|
||||
IO.Iteratee
|
||||
^^^^^^^^^^^
|
||||
|
||||
Included with Akka's IO support is a basic implementation of ``Iteratee``\s. ``Iteratee``\s are an effective way of handling a stream of data without needing to wait for all the data to arrive. This is especially useful when dealing with non blocking IO since we will usually receive data in chunks which may not include enough information to process, or it may contain much more data than we currently need.
|
||||
|
||||
This ``Iteratee`` implementation is much more basic than what is usually found. There is only support for ``ByteString`` input, and enumerators aren't used. The reason for this limited implementation is to reduce the amount of explicit type signatures needed and to keep things simple. It is important to note that Akka's ``Iteratee``\s are completely optional, incoming data can be handled in any way, including other ``Iteratee`` libraries.
|
||||
|
||||
``Iteratee``\s work by processing the data that it is given and returning either the result (with any unused input) or a continuation if more input is needed. They are monadic, so methods like ``flatMap`` can be used to pass the result of an ``Iteratee`` to another.
|
||||
|
||||
The basic ``Iteratee``\s included in the IO support can all be found in the ScalaDoc under ``akka.actor.IO``, and some of them are covered in the example below.
|
||||
|
||||
Examples
|
||||
--------
|
||||
|
||||
Http Server
|
||||
^^^^^^^^^^^
|
||||
|
||||
This example will create a simple high performance HTTP server. We begin with our imports:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: imports
|
||||
|
||||
Some commonly used constants:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: constants
|
||||
|
||||
And case classes to hold the resulting request:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: request-class
|
||||
|
||||
Now for our first ``Iteratee``. There are 3 main sections of a HTTP request: the request line, the headers, and an optional body. The main request ``Iteratee`` handles each section separately:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-request
|
||||
|
||||
In the above code ``readRequest`` takes the results of 3 different ``Iteratees`` (``readRequestLine``, ``readHeaders``, ``readBody``) and combines them into a single ``Request`` object. ``readRequestLine`` actually returns a tuple, so we extract it's individual components. ``readBody`` depends on values contained within the header section, so we must pass those to the method.
|
||||
|
||||
The request line has 3 parts to it: the HTTP method, the requested URI, and the HTTP version. The parts are separated by a single space, and the entire request line ends with a ``CRLF``.
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-request-line
|
||||
|
||||
Reading the request method is simple as it is a single string ending in a space. The simple ``Iteratee`` that performs this is ``IO.takeUntil(delimiter: ByteString): Iteratee[ByteString]``. It keeps consuming input until the specified delimiter is found. Reading the HTTP version is also a simple string that ends with a ``CRLF``.
|
||||
|
||||
The ``ascii`` method is a helper that takes a ``ByteString`` and parses it as a ``US-ASCII`` ``String``.
|
||||
|
||||
Reading the request URI is a bit more complicated because we want to parse the individual components of the URI instead of just returning a simple string:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-request-uri
|
||||
|
||||
For this example we are only interested in handling absolute paths. To detect if we the URI is an absolute path we use ``IO.peek(length: Int): Iteratee[ByteString]``, which returns a ``ByteString`` of the request length but doesn't actually consume the input. We peek at the next bit of input and see if it matches our ``PATH`` constant (defined above as ``ByteString("/")``). If it doesn't match we throw an error, but for a more robust solution we would want to handle other valid URIs.
|
||||
|
||||
Next we handle the path itself:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-path
|
||||
|
||||
The ``step`` method is a recursive method that takes a ``List`` of the accumulated path segments. It first checks if the remaining input starts with the ``PATH`` constant, and if it does, it drops that input, and returns the ``readUriPart`` ``Iteratee`` which has it's result added to the path segment accumulator and the ``step`` method is run again.
|
||||
|
||||
If after reading in a path segment the next input does not start with a path, we reverse the accumulated segments and return it (dropping the last segment if it is blank).
|
||||
|
||||
Following the path we read in the query (if it exists):
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-query
|
||||
|
||||
It is much simpler than reading the path since we aren't doing any parsing of the query since there is no standard format of the query string.
|
||||
|
||||
Both the path and query used the ``readUriPart`` ``Iteratee``, which is next:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-uri-part
|
||||
|
||||
Here we have several ``Set``\s that contain valid characters pulled from the URI spec. The ``readUriPart`` method takes a ``Set`` of valid characters (already mapped to ``Byte``\s) and will continue to match characters until it reaches on that is not part of the ``Set``. If it is a percent encoded character then that is handled as a valid character and processing continues, or else we are done collecting this part of the URI.
|
||||
|
||||
Headers are next:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-headers
|
||||
|
||||
And if applicable, we read in the message body:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: read-body
|
||||
|
||||
Finally we get to the actual ``Actor``:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: actor
|
||||
|
||||
And it's companion object:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: actor-companion
|
||||
|
||||
And the OKResponse:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: ok-response
|
||||
|
||||
A ``main`` method to start everything up:
|
||||
|
||||
.. includecode:: code/docs/io/HTTPServer.scala
|
||||
:include: main
|
||||
|
|
@ -26,10 +26,6 @@ asynchronous. The API is meant to be a solid foundation for the implementation
|
|||
of network protocols and building higher abstractions; it is not meant to be a
|
||||
full-service high-level NIO wrapper for end users.
|
||||
|
||||
.. note::
|
||||
|
||||
The old I/O implementation has been deprecated and its documentation has been moved: :ref:`io-scala-old`
|
||||
|
||||
Terminology, Concepts
|
||||
---------------------
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue