More Scala resources
Coroutines
I tend to think of a coroutine as a communicating process. A producer is a coroutine that emits values. A consumer is a coroutine that receives values. A transducer is a coroutine that both emits and receives. It's easy to represent coroutines as threads in Scala:
trait Coroutine extends Runnable {
def start () {
val myThread = new Thread(this) ;
myThread.start() ;
}
}
(It's also possible to use continuations, which Scala now has, to build coroutines.)
A simple (only mildly inefficient) implementation of consumers,
producers and transducers uses thread-safe queues to buffer inputs and
outputs.
Producers call put() inside an infinite loop in their run() method when they have a value to emit:
/**
A O-producer is a coroutine that produces
type-O objects for consumption by other
coroutines.
*/
trait Producer[O] extends Coroutine {
private val outputs =
new ArrayBlockingQueue[O] (1024) ;
/**
Called by the coroutine's run
method when it has produced a new output.
*/
protected def put (output : O) {
outputs.put(output) ;
}
/**
Called by an upstream consumer when it
wants a new value from this coroutine.
*/
def next () : O = {
outputs.take() ;
}
/**
Composes this producing coroutine
with a transducing coroutine.
@return A fused producing coroutine.
*/
def ==>[O2] (t : Transducer[O,O2])
: Producer[O2] = {
val that = this ;
new Producer[O2] {
def run () {
while (true) {
val o = that.next() ;
t.accept(o);
put(t.next()) ;
}
}
override def start () {
that.start() ;
t.start() ;
super.start() ;
}
}
}
/**
Composes this producing coroutine
with a consuming coroutine.
@return A fused coroutine.
*/
def ==> (consumer : Consumer[O])
: Coroutine = {
val that = this ;
new Coroutine {
def run {
while (true) {
val o = that.next() ;
consumer.accept(o) ;
}
}
override def start {
that.start() ;
consumer.start() ;
super.start() ;
}
}
}
}
Consumers call get():
/**
An I-consumer is a coroutine that consumes
type-I objects.
*/
trait Consumer[I] extends Coroutine {
private val inputs =
new ArrayBlockingQueue[I] (1024) ;
/**
Called when an external I-producer
has a new input to provide.
*/
def accept (input : I) {
inputs.put(input) ;
}
/**
Called by the coroutine itself
when it needs the next input.
*/
protected def get () : I = {
inputs.take() ;
}
}
Transducers are literally both a consumer and a producer:
/** An I,O-transducer consumes type-I objects and produces type-O objects. */ trait Transducer[I,O] extends Consumer[I] with Producer[O]
One magic line
Using coroutines, the punchline in the web server becomes a single line:
listen ==> connect ==> process ==> reply
where
listenis a producer that listens on a port and emits connected sockets;connectis a transducer that receives newly connected sockets, wraps them into a connection and then extracts HTTP requests (using non-blocking I/O);processis a transducer that consumes HTTP requests and produces replies; andreplyis a consumer that sends replies back (using non-blocking I/O).
The value of this one-line expression is a web server.
Web server source code
Download a zip with source, README and a Makefile.
/*
M2HTTPD: A tiny yet pipelined, non-blocking, extensible HTTPD.
Author: Matthew Might
Site: http://matt.might.net/
M2HTTPD is implemented using a pipeline of transducers.
A transducer is a kind of coroutine (a communicating process) that
consumes upstream inputs and sends outputs downstream.
M2HTTPD has four transducers chained together into a pipeline:
(1) Phase 1 listens for incoming connections and passes
connected sockets downstream.
(2) Phase 2 wraps a socket in a connection object and parses
an HTTP request out of it.
(3) Phase 3 routes a request to the appropriate request handler,
and if the handler produces a reply, the reply is sent
downstream.
(4) Phase 4 consumes replies and pushes them onto the network.
By slicing a task into phases, it is possible to operate on multiple
tasks at the same time, thereby increasing the throughput of the
overall system.
No blocking IO calls are used anywhere, which makes some kinds of
attacks against the HTTPD (e.g. SYN floods) ineffective.
*/
import java.net.InetAddress ;
import java.net.InetSocketAddress ;
import java.net.URLDecoder ;
import java.nio.ByteBuffer ;
import java.nio.channels.SocketChannel ;
import java.nio.channels.FileChannel ;
import java.nio.channels.Selector ;
import java.nio.channels.spi.SelectorProvider ;
import java.nio.channels.SelectionKey ;
import java.io.File ;
import java.io.FileInputStream ;
import java.io.FileWriter ;
import java.io.FileReader ;
import java.io.BufferedReader ;
import java.io.BufferedWriter ;
import java.io.IOException ;
import java.util.concurrent.ConcurrentHashMap ;
/* Generic utilities */
/**
Provides debugging utilities.
*/
object Debugger {
var isActive = true
def debugln(o : Object) {
if (this isActive)
System.err.println (o) ;
}
}
import Debugger._ ;
/**
Provides transparent compatibility with Java iterators.
*/
object JavaImplicits {
implicit def javaIteratorToScalaIterator[A](it : java.util.Iterator[A]) =
new scala.collection.jcl.MutableIterator.Wrapper(it) ;
}
import JavaImplicits._ ;
/**
Typed data has a mime type and can be rendered as a byte buffer.
*/
trait TypedData {
def contentType : String ;
def asByteBuffer : ByteBuffer ;
}
/**
Raw typed data is a contet type plus a byte buffer.
*/
class RawTypedData(val contentType : String, byteBuffer : ByteBuffer) extends TypedData {
def asByteBuffer = byteBuffer.asReadOnlyBuffer
}
/**
Transparently converts Scala data into typed data.
*/
object TypedDataImplicits {
implicit def stringToTypedData(string : String) = new TypedData {
def contentType = "text/html"
def asByteBuffer = ByteBuffer.wrap(string.getBytes("UTF-8"))
}
}
import TypedDataImplicits._ ;
/**
Provides convenience functions for manipulating times.
*/
object Time {
/**
The current time is milliseconds.
*/
def now () : Long = new java.util.Date().getTime()
private val gmtFormat =
new java.text.SimpleDateFormat("E, d MMM yyyy HH:mm:ss 'GMT'",
java.util.Locale.US);
gmtFormat.setTimeZone(java.util.TimeZone.getTimeZone("GMT"));
/**
The current time rendered in the format required by the HTTP specification.
*/
def asHTTPDate() : String = gmtFormat.format(new java.util.Date());
/**
A time rendered in the format required by the HTTP specification.
*/
def asHTTPDate(time : Long) = gmtFormat.format(time) ;
/**
A time a now + offset ms into the future in the format required by the HTTP specification.
Useful for setting cache expiration times.
*/
def asFutureHTTPDate(offset : Long) =
gmtFormat.format(new java.util.Date().getTime() + offset);
}
/**
A coroutine is a process (in this case a thread) that communicates with other coroutines.
*/
trait Coroutine extends Runnable {
def start () {
val myThread = new Thread(this) ;
myThread.start() ;
}
}
/**
A O-producer is a coroutine that produces type-O objects for consumption by other coroutines.
*/
trait Producer[O] extends Coroutine {
private val outputs =
new java.util.concurrent.ArrayBlockingQueue[O] (1024) ;
/**
Called by the coroutine's <code>run</code> method when it has produced a new output.
*/
protected def put (output : O) {
outputs.put(output) ;
}
/**
Called by an upstream consumer when it wants a new value from this coroutine.
*/
def next () : O = {
outputs.take() ;
}
/**
Composes this producing coroutine with a transducing coroutine.
@return A fused producing coroutine.
*/
def ==>[O2] (t : Transducer[O,O2]) : Producer[O2] = {
val that = this ;
new Producer[O2] {
def run () {
while (true) {
val o = that.next() ;
t.accept(o);
put(t.next()) ;
}
}
override def start () {
that.start() ;
t.start() ;
super.start() ;
}
}
}
/**
Composes this producing coroutine with a consuming coroutine.
@return A fused coroutine.
*/
def ==> (consumer : Consumer[O]) : Coroutine = {
val that = this ;
new Coroutine {
def run {
while (true) {
val o = that.next() ;
consumer.accept(o) ;
}
}
override def start {
that.start() ;
consumer.start() ;
super.start() ;
}
}
}
}
/**
An I-consumer is a coroutine that consumes type-I objects.
*/
trait Consumer[I] extends Coroutine {
private val inputs =
new java.util.concurrent.ArrayBlockingQueue[I] (1024) ;
/**
Called when an external I-producer has a new input to provide.
*/
def accept (input : I) {
inputs.put(input) ;
}
/**
Called by the coroutine itself when it needs the next input.
*/
protected def get () : I = {
inputs.take() ;
}
}
/**
An I,O-transducer consumes type-I objects and produces type-O objects.
*/
trait Transducer[I,O] extends Consumer[I] with Producer[O]
/**
Manages a stable of threads, ready to execute tasks concurrently.
*/
class ThreadFarm (val queueSize : Int, val numberThreads : Int) {
private val tasks =
new java.util.concurrent.ArrayBlockingQueue[() => Unit] (queueSize) ;
/**
Each work pulls a task and executes it, looping forever.
*/
private object Worker extends Runnable {
def run () {
while (true) {
try {
val task = tasks.take()
task()
} catch {
case (ex : Exception) => {
ex.printStackTrace() ;
}
}
}
}
}
/**
Starts the stable.
*/
def start () {
for (i <- 1 to numberThreads) {
val workerThread = new Thread(Worker)
workerThread.start()
}
}
/**
Accepts an action to be run concurrently.
*/
def run (action : => Unit) {
tasks.put(() => action)
}
/**
Accepts a taks to be run concurrently.
*/
def addTask (task : () => Unit) {
tasks.put(task)
}
}
/* M2HTTPD */
/**
Represents an HTTP request method.
*/
trait HTTPMethod
case object POST extends HTTPMethod
case object GET extends HTTPMethod
/**
A producing coroutine that listens to a port and emits
connected sockets.
*/
class PortListener (val localAddress : InetAddress,
val port : Int) extends Producer[SocketChannel]
{
// Non-blocking server socket:
private var listener : java.nio.channels.ServerSocketChannel = null ;
/**
Opens the port and loops forever, emitting newly connected sockets.
*/
def run () {
debugln ("PortListener initializing...") ;
var client : java.nio.channels.SocketChannel = null ;
this.listener = java.nio.channels.ServerSocketChannel.open() ;
listener.socket().bind(new java.net.InetSocketAddress(localAddress,
port)) ;
listener.configureBlocking(true) ;
while (true) {
client = listener.accept() ;
if (client != null) {
client.socket() ;
client.configureBlocking(false) ;
debugln (" * Accepted client") ;
put(client) ;
}
}
}
}
/**
Represents an open HTTP connection coming from a client.
*/
class ClientConnection (val httpd : M2HTTPD,
val socketChannel : SocketChannel)
{
private var _isOpen = true
/**
Is this connection currentnly open?
*/
def isOpen = _isOpen
private var requestParser : HTTPRequestParser = new HTTPRequestParser
/**
Closes the connection.
*/
def close () {
_isOpen = false ;
socketChannel.close() ;
}
/**
Has this connection parsed a request?
*/
def hasRequest = (requestParser != null) && requestParser.parsed
/**
Produces the request parsed by this parser.
*/
def takeRequest () : Request = {
val req = requestParser.toRequest(this)
// TODO: Needs to be a seconday request parser
// for Connection: Keep-Alive
requestParser = null ;
req
}
/**
Stores input during reads.
*/
private val readBuffer = ByteBuffer.allocate(8192);
/**
Reads more input, if any, and passes it to the http request parser.
*/
def readMoreInput () {
debugln (" ** Connection has input.") ;
if (!_isOpen)
throw new IOException("Cannot read a closed connection!") ;
this.readBuffer.clear() ;
var numRead = -1 ;
try {
numRead = socketChannel.read(this.readBuffer) ;
} catch {
case (ioe : IOException) => {
this.close()
throw ioe
}
}
if (numRead == -1) {
this.close()
throw new IOException("Remote entity shut down socket (cleanly).") ;
}
val bytes = this.readBuffer.array() ;
for (i <- 0 until numRead) {
requestParser.consume(bytes(i).asInstanceOf[Char]) ;
}
}
/**
A list of buffers that need to be written.
*/
private val writeMutex = new Object()
private var writeBufferList : List[ByteBuffer] = List()
/**
Queues a reply to send to this connection.
<code>flushWriteBuffers</code> must be called until it returns true to
guarantee the reply has been sent.
*/
def send (reply : Reply) {
writeBufferList = reply.headerByteBuffer :: reply.dataByteBuffer :: writeBufferList
}
/**
Sends as much data in its buffer queue as it can without blocking.
@return True if all data has been sent; false if not.
*/
def flushWriteBuffers() : Boolean = {
if (!_isOpen) {
System.err.println ("ERROR: Attempt to write to closed channel.")
return true ;
}
if (!writeBufferList.isEmpty) {
if (writeBufferList.head.remaining() > 0) {
try {
socketChannel.write(writeBufferList.head) ;
} catch {
case (ioe : IOException) => {
this.close() ;
throw ioe
}
}
}
if (writeBufferList.head.remaining() == 0) {
writeBufferList = writeBufferList.tail
}
}
return writeBufferList isEmpty ;
}
}
/**
A continuation-based HTTP request parser.
The parser consumes one character at a time,
which means that the parsing process can be suspended at any time.
At the moment, this does not support Keep-Alive HTTP connections.
*/
class HTTPRequestParser {
/**
Indicates the current position of the parser.
*/
private trait ParsingState ;
private case class FirstLine(val line : String) extends ParsingState
private case class HeaderLine(val line : String) extends ParsingState
private case class NextLine(val line : String) extends ParsingState
private case class Data(val data : StringBuilder) extends ParsingState
private case object End extends ParsingState
private case object EndHeaders extends ParsingState
/**
Marks the current state of the parser.
*/
private var state : ParsingState = FirstLine("") ;
/* Components of the request. */
private var method : HTTPMethod = null ;
private var resource : String = null ;
// OPTIMIZATION: Use a faster map.
private var headers : scala.collection.immutable.Map[String,String] = Map() ;
private var data : String = null ;
private var contentLength : Int = -1 ;
/**
Has a complete request been parsed?
*/
def parsed = state == End || ((method == GET) && state == EndHeaders)
private def processFirstLine(line : String) {
val parts = line.split(" ") ;
parts(0) match {
case "GET" => this.method = GET ;
case "POST" => this.method = POST ;
case _ => throw new Exception("Unknown HTTP method: " + parts(0))
}
resource = parts(1) ;
}
private def processHeaderLine(line : String) {
val parts = line.split(": ") ;
headers = (headers(parts(0)) = parts(1)) ;
headers get ("Content-Length") match {
case Some(l) => {
contentLength = Integer.parseInt(l)
}
case None => {}
}
}
/**
Update the state of the parser with the next character.
*/
def consume (c : Char) {
state =
(state,c) match {
case (FirstLine(ln),'\r'|'\n') => { processFirstLine(ln) ; NextLine(""+c) }
case (FirstLine(ln),_) => FirstLine(ln + c)
case (NextLine("\n"),'\n') => EndHeaders
case (NextLine("\r\n\r"),'\n') => EndHeaders
case (NextLine("\r"),'\n') => NextLine("\r\n")
case (NextLine("\r\n"),'\r') => NextLine("\r\n\r")
case (NextLine(_), c) => HeaderLine(String.valueOf(c))
case (HeaderLine(ln),'\r'|'\n') => { processHeaderLine(ln) ; NextLine(""+c) }
case (HeaderLine(ln),_) => HeaderLine(ln + c)
case (EndHeaders,c) if contentLength > 1 => {
val sb = new StringBuilder() ;
sb.append(c) ;
Data(sb) ;
}
case (Data(sb),c) if contentLength > (sb.length + 1) => {
sb.append(c) ;
Data(sb)
}
case (Data(sb),c) => {
sb.append(c)
this.data = sb.toString
End
}
case _ => throw new Exception(state + "/" + c + "\n\n" + headers)
}
}
/**
If a request is completed, produce the Request object.
*/
def toRequest(conn : ClientConnection) : Request = {
if (this.parsed)
new Request(conn,method,resource,headers,data)
else
throw new Exception("HTTP request not completed!")
}
}
/**
A coroutine that consumes client connections and produces HTTP requests from them.
*/
class ConnectionManager (val httpd : M2HTTPD) extends Transducer[SocketChannel,Request] {
private val farm = new ThreadFarm(1024,10)
override def start () {
super.start()
farm.start()
}
private object ConnectionSelector extends Runnable {
private val newConnections =
new java.util.concurrent.LinkedBlockingQueue[ClientConnection] () ;
// BUG: Once got a null pointer exception here.
// Probably a race condition somewhere in the runtime lib.
/*
Exception in thread "Thread-3" java.lang.NullPointerException
at sun.nio.ch.Util.atBugLevel(Util.java:326)
at sun.nio.ch.SelectorImpl.<init>(SelectorImpl.java:40)
at sun.nio.ch.KQueueSelectorImpl.<init>(KQueueSelectorImpl.java:46)
at sun.nio.ch.KQueueSelectorProvider.openSelector(KQueueSelectorProvider.java:20)
at ConnectionManager$ConnectionSelector$.<init>(M2HTTPD.scala:562)
*/
private val selector : Selector =
SelectorProvider.provider().openSelector() ;
def add(conn : ClientConnection) {
newConnections.put(conn);
this.selector.wakeup() ;
}
def start () {
val thread = new Thread(this) ;
thread.start () ;
}
def run () {
debugln("Connection Selector initialized...")
while (true) {
selector.select() ;
val selectedKeys = selector.selectedKeys.iterator() ;
debugln(" ** Reading keys selected.") ;
while (selectedKeys.hasNext()) {
val key = selectedKeys.next().asInstanceOf[SelectionKey] ;
selectedKeys.remove() ;
if (key.isValid()) {
val conn = key.attachment().asInstanceOf[ClientConnection] ;
// Remove the key so we don't immediately loop around to
// race on the same connection.
key.cancel() ;
farm run {
try {
// Read whatever is available.
conn.readMoreInput() ;
// Return this connection to the read selector.
add(conn) ;
} catch {
case (ioe : IOException) => {
// Socket shut down.
key.cancel() ;
}
}
// Was enough input read to complete a request?
if (conn.hasRequest) {
put(conn.takeRequest()) ;
}
}
}
}
// Add queued up connections before looping back around:
val conns = new java.util.Vector[ClientConnection]
newConnections.drainTo(conns) ;
for (conn <- conns.iterator()) {
if (conn.isOpen)
conn.socketChannel.register(this.selector, SelectionKey.OP_READ, conn) ;
}
}
}
}
def run () {
debugln ("Connection Manager initializing...");
ConnectionSelector.start() ;
while (true) {
val socket = get() ;
val conn = new ClientConnection(httpd,socket)
debugln (" * Got a requesting connection.") ;
ConnectionSelector.add(conn) ;
}
}
}
/**
Represents an incoming HTTP request.
*/
class Request (val connection : ClientConnection,
val method : HTTPMethod,
val resource : String,
val headers : Map[String,String],
val data : String)
{
/**
Contains user-definable attributes.
For example, a session-handling request handler could insert session data here.
*/
val attributes = scala.collection.mutable.HashMap[String,Object]()
private lazy val resourceParts = resource.split("\\?")
/**
Contains the path to the requested resource.
*/
lazy val path = resourceParts(0) ;
/**
Contains the complete query string for the request.
*/
lazy val queryString = if (resourceParts.length == 2) resourceParts(1) else "" ;
/**
Contains the components of the path to the requested resource.
*/
lazy val pathComponents = path.split("//+")
/**
@return A map representing a query string of the form "param1=value1¶m2=value2&..."
*/
private def queryStringToMap (queryString : String) : Map[String,String] = {
val queryStrings = queryString.split("&")
if (queryString.length > 0) {
Map() ++ (for (param <- queryStrings) yield {
val keyValue = param.split("=") ;
keyValue match {
case Array(key,value) => (URLDecoder.decode(key,"UTF-8"),
URLDecoder.decode(value,"UTF-8"))
case Array(key) => (URLDecoder.decode(key,"UTF-8"),"true")
}
})
} else {
return Map()
}
}
/**
Maps a parameter name to its value.
If the parameter was not assigned, its value is "true".
*/
lazy val query : Map[String,String] = queryStringToMap(queryString)
/**
Maps a POST parameter to its value.
*/
lazy val posts : Map[String,String] =
(headers get "Content-Type") match {
case Some("application/x-www-form-urlencoded") => queryStringToMap(data)
case None => Map()
}
/**
@return A debug-friendly representation of this procedure.
*/
override def toString : String = {
"Method: " + method + "\n" +
"Resource: " + resource + "\n" +
"Headers: " + headers + "\n" +
"Query: " + query + "\n" +
"Post: " + posts + "\n" +
"Data: " + data + "\n" ;
}
}
/**
A coroutine that consumes HTTP requests and produces HTTP replies.
*/
class RequestProcessor extends Transducer[Request,Reply] {
private val farm = new ThreadFarm(1024,10)
override def start () {
super.start()
farm.start()
}
/**
The host router determines which request handler should be used for the
host specified in the request.
The default hostRouter produces 404s for every request.
*/
var hostRouter : HostRouter =
EmptyHostRouter ;
/**
Consumes requests and farms them to request handlers.
*/
def run () {
while (true) {
val req = get() ;
debugln(req) ;
farm run {
var handler : RequestHandler = null
try {
handler = hostRouter(req.headers("Host"))
// If the handler produces a reply, pass it on.
handler(req,None) match {
case Some (reply) => put(reply)
case None => ()
}
} catch {
case ex => {
System.err.println("error:\n" + ex)
val reply = new Reply(req,500,"Internal server error")
put(reply)
}
}
}
}
}
}
/**
A coroutine that consumes replies and sends them off without blocking.
*/
class ReplySender extends Consumer[Reply] {
private val farm = new ThreadFarm(1024,10)
override def start () {
super.start()
farm.start()
}
private object ConnectionSelector extends Runnable {
private val newConnections =
new java.util.concurrent.LinkedBlockingQueue[ClientConnection] () ;
private val selector : Selector =
SelectorProvider.provider().openSelector() ;
def add(conn : ClientConnection) {
newConnections.put(conn);
this.selector.wakeup() ;
}
def start () {
val thread = new Thread(this) ;
thread.start () ;
}
def run () {
debugln("Reply Sender Selector initialized...")
while (true) {
selector.select() ;
val selectedKeys = selector.selectedKeys.iterator() ;
debugln(" ** Writing keys selected.") ;
while (selectedKeys.hasNext()) {
val key = selectedKeys.next().asInstanceOf[SelectionKey] ;
selectedKeys.remove() ;
if (key.isValid()) {
val conn = key.attachment().asInstanceOf[ClientConnection] ;
key.cancel() ;
if (!conn.isOpen) {
conn.close() ;
}
farm run {
try {
debugln(" ** Sending reply!") ;
val finished = conn.flushWriteBuffers() ;
if (!finished) {
add(conn) ;
} else {
conn.close() ;
}
} catch {
case (ioe : IOException) => {
// Socket shut down.
key.cancel() ;
}
}
}
}
}
// Adding new connections.
val conns = new java.util.Vector[ClientConnection]
newConnections.drainTo(conns) ;
for (conn <- conns.iterator()) {
conn.socketChannel.register(this.selector, SelectionKey.OP_WRITE, conn) ;
}
}
}
}
def run () {
debugln ("Reply Sender initializing...");
ConnectionSelector.start() ;
while (true) {
val reply = get() ;
val conn = reply.req.connection ;
conn.send(reply) ;
debugln (" * Sending a reply.") ;
ConnectionSelector.add(conn) ;
}
}
}
/**
Handles HTTP requests.
Request handlers are designed to compose, to support, for example,
routing handlers or session manager handlers or reformatting
handlers.
*/
abstract class RequestHandler {
/**
If a request has already been handled by other handlers,
their composed reply, if any, is provided.
*/
def apply (req : Request, reply : Option[Reply]) : Option[Reply] ;
/**
Composes two request handlers.
*/
def ==> (handler : RequestHandler) {
var that = this ;
new RequestHandler {
def apply (req : Request, reply : Option[Reply]) : Option[Reply] = {
val firstReply = that(req,reply)
handler(req,firstReply)
}
}
}
}
/**
Always sends 404 File Not Found.
*/
object EmptyRequestHandler extends RequestHandler {
def apply(req : Request, reply : Option[Reply]) : Option[Reply] = {
val notFoundReply = new Reply(req,404,"File not found")
Some(notFoundReply)
}
}
/**
Serves as a (very) basic file server.
*/
class SimpleFileRequestHandler (var docRoot : String) extends RequestHandler {
def apply(req : Request, reply : Option[Reply]) : Option[Reply] = {
val filePath = docRoot + "/" + req.resource ;
val file = new File(filePath)
if (!file.exists)
// Send a 404.
return EmptyRequestHandler(req,reply)
val fileChannel = (new java.io.RandomAccessFile(filePath, "r")).getChannel() ;
val size = fileChannel.size() ;
val byteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY,0,size) ;
val mimeType = DefaultMIMETypeMap(req.resource)
return Some(new Reply(req, 200, new RawTypedData(mimeType, byteBuffer)))
}
}
/**
A host router selects a request handler based on the virtual host.
*/
abstract class HostRouter {
val fallback : HostRouter = EmptyHostRouter ;
def apply (host : String) : RequestHandler ;
}
/**
Always returns the <code>EmptyRequestHandler</code>.
*/
object EmptyHostRouter extends HostRouter {
def apply (host : String) : RequestHandler = EmptyRequestHandler ;
}
/**
Always returns the supplied request handler, regardless of host.
*/
class DefaultHostRouter (defaultHandler : RequestHandler) extends HostRouter {
def apply (host : String) = defaultHandler
}
/**
Maintains a hash table of virtual hosts and their associated request handlers.
*/
class HashtableHostRouter (override val fallback : HostRouter) extends HostRouter {
def this () = this (EmptyHostRouter)
private val tableMutex = new Object()
@volatile private var table =
scala.collection.immutable.HashMap[String,RequestHandler]()
/**
@return The request handler if it exists, or else looks it up in the fallback host router.
*/
def apply(host : String) : RequestHandler = try {
table.getOrElse (host, fallback(host))
}
/**
Installs a request handler at the specified host.
*/
def update (host : String, handler : RequestHandler) {
tableMutex synchronized {
table = (table(host) = handler)
}
}
}
/**
A reply to an HTTP request.
*/
case class Reply (val req : Request,
var code : Int,
var data : TypedData)
{
def this (req : Request) = this (req,404,"File not found")
/**
Maps a reply header to its value.
*/
val headers = scala.collection.mutable.HashMap[String,String]() ;
headers("Server") = "Apache/1.3.3.7 (Unix) (Red-Hat/Linux)" ;
headers("Connection") = "Close" ;
/**
Renders the headers for an HTTP/1.1-compliant reply.
*/
private def headerString () : String = {
val s = new StringBuilder
s.append("HTTP/1.1 " + code + " " + HTTPCodeName(code)) ;
for ((h,v) <- headers) {
s.append(h + ": " + v + "\r\n")
}
s.append("\r\n") ;
return s.toString
}
/**
Contains the data to be sent to the client.
*/
lazy val dataByteBuffer : ByteBuffer =
data.asByteBuffer
/**
Contains the header to be sent to the client.
*/
lazy val headerByteBuffer : ByteBuffer = {
val length = dataByteBuffer.capacity
if (data.contentType != "unknown/unknown")
// If you don't know the content type, let the browser guess.
headers("Content-Type") = data.contentType
headers("Content-Length") = String.valueOf(length)
headers("Date") = Time.asHTTPDate()
val head = headerString ()
ByteBuffer.wrap(head.getBytes("UTF-8"))
}
}
/**
Maps a filename/extension to its MIME type.
*/
class MIMETypeMap {
private val map = scala.collection.mutable.HashMap[String,String]()
map("html") = "text/html" ;
map("txt") = "text/plain" ;
map("css") = "text/css" ;
map("js") = "application/javascript" ;
map("png") = "image/png" ;
map("jpg") = "image/jpeg" ;
map("gif") = "image/gif" ;
def apply (fileName : String) : String = {
val parts = fileName.split("[.]")
map get (parts(parts.length-1)) match {
case Some(ty) => ty
case None => "unknown/unknown"
}
}
}
/**
Maps filenames/extensions to a best guess of their MIME type.
*/
object DefaultMIMETypeMap extends MIMETypeMap
/**
Converts HTTP status codes into their corresponding string.
For example, 200 becomes "OK" and 404 becomes "Not Found".
*/
object HTTPCodeName {
private var table = new Array[String](1000)
table(200) = "OK"
table(400) = "Bad Request"
table(404) = "Not Found"
def apply (code : Int) : String = table(code)
}
/**
An extensible, piplined HTTPD that uses non-blocking IO.
*/
class M2HTTPD {
/**
The port on which the HTTPD listens.
By default, 1701.
*/
var port = 1701 ;
/**
The address to which the HTTPD listens.
By default, null. (null means list to all local addresses.)
*/
var localAddress : InetAddress = null ;
/**
A coroutine that listens to a port to produce sockets.
*/
val listen = new PortListener (localAddress,port) ;
/**
A coroutine that turns sockets into HTTP requests.
*/
val connect = new ConnectionManager(this) ;
/**
A coroutine that routes and processes requests.
*/
val process = new RequestProcessor ;
/**
A coroutine that sends replies
*/
val reply = new ReplySender ;
/**
A pipelined coroutine that ties together all of the stages of the HTTPD.
*/
val system = listen ==> connect ==> process ==> reply
/**
Starts up the HTTPD.
*/
def start () {
system.start()
}
}
/* Demo servers */
/**
A driver that starts the M2HTTPD with the default (all-404) request handler.
*/
object EmptyHTTPD {
def main (args : Array[String]) {
debugln("Default M2HTTPD running...")
val httpd = new M2HTTPD ;
httpd.start()
}
}
/**
A driver that starts the M2HTTPD with the simple file server request handler.
*/
object SimpleHTTPD {
def main (args : Array[String]) {
debugln("Simple M2HTTPD running...")
val httpd = new M2HTTPD ;
// By default, serve from ./www/ in the current directory.
var docRoot = "./www/" ;
args match {
case Array(newRoot) => docRoot = newRoot
case _ => ()
}
debugln("Serving from " + docRoot) ;
// If you want to take control of the HTTPD, then take control of
// the hostRouter, and create a request handler:
httpd.process.hostRouter =
new DefaultHostRouter(new SimpleFileRequestHandler(docRoot)) ;
httpd.start()
}
}
Twitter: @mattmight
Instagram: @mattmight
LinkedIn: matthewmight
Mastodon: @mattmight@mathstodon.xyz
Sub-reddit: /r/mattmight