root/blogbird/src/main/scala/org/aiotrade/httpd/ClientConnection.scala @ 266:99aa5be7e923

package org.aiotrade.httpd

import java.io.IOException
import java.nio.ByteBuffer
import java.nio.channels.SocketChannel

import org.aiotrade.lib.util.io.ChannelListener
import org.aiotrade.lib.util.io.SelectActor.Read
import org.aiotrade.lib.util.io.SelectActor.Write
import scala.actors.Actor
import Debugger._

/**
 * Represents an open HTTP connection coming from a client.
 */
class ClientConnection(val channel: SocketChannel) extends Actor with ChannelListener {
  private var _isOpen = true
  private var requestParser = new HttpRequestParser

  start
  println("clientConnection started")
  
  /**
   * Is this connection currentnly open?
   */
  def isOpen = _isOpen && channel.isOpen && channel.isConnected

  /**
   * Closes the connection.
   */
  def close {
    _isOpen = false
    channel.close
    super.exit
  }

  def act = loop {
    react {
      case Read(sender) =>
        try {
          val hasRequest = readMoreInput

          // add back this connection to the read selector.
          sender.addListener(this)

          // was enough input read to complete a request?
          if (hasRequest) {
            ConnectionManager.selectActor.produce(takeRequest)
          }
        } catch {case ex: IOException => debugln(ex.getMessage)}
        
      case Write(sender) =>
        debugln(" ** Sending reply!")
        try {
          val finished = flushWriteBuffers
          if (finished) {
            close
          } else {
            // add back this connection to the write selector.
            sender.addListener(this)
          }
        } catch {case ex: IOException => debugln(ex.getMessage)}
    }
  }

  /**
   * 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 readBuf = ByteBuffer.allocate(8192)

  /**
   * Reads more input, if any, and passes it to the http request parser.
   * 
   * @return True if all data has been sent; false if not.
   */
  def readMoreInput: Boolean = {
    debugln(" ** Connection has input.")

    if (!_isOpen)
      throw new IOException("Cannot read a closed connection!")

    readBuf.clear
    var numRead = -1
    try {
      numRead = channel.read(readBuf)
    } catch {
      case ioe: IOException =>
        debugln(ioe.getMessage)
        close
        throw ioe
    }

    if (numRead == -1) {
      close
      debugln("Remote entity shut down socket (cleanly).")
      throw new IOException("Remote entity shut down socket (cleanly).")
    }

    val bytes = readBuf.array
    var i = 0
    while (i < numRead) {
      requestParser.consume(bytes(i).asInstanceOf[Char])
      i += 1
    }
    
    hasRequest
  }

  /**
   * A list of buffers that need to be written.
   */
  private val writeMutex = new Object()
  private var writeBuffers = List[ByteBuffer]()


  /**
   * 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) {
    writeBuffers = reply.headerByteBuffer :: reply.dataByteBuffer :: writeBuffers
  }


  /**
   * 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 (!writeBuffers.isEmpty) {
      if (writeBuffers.head.remaining > 0) {
        try {
          channel.write(writeBuffers.head)
        } catch {
          case ioe: IOException =>
            close
            throw ioe
        }
      }
      if (writeBuffers.head.remaining == 0) {
        writeBuffers = writeBuffers.tail
      }
    }

    writeBuffers isEmpty
  }
}


/**
 * Represents an HTTP request method.
 */
trait HttpMethod

case object POST extends HttpMethod
case object GET  extends HttpMethod

/**
 * 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 State
  private case class FirstLine(line: String) extends State
  private case class HeaderLine(line: String) extends State
  private case class NextLine(line: String) extends State
  private case class Data(data: StringBuilder) extends State
  private case object End extends State
  private case object EndHeaders extends State

  /**
   * Marks the current state of the parser.
   */
  private var state: State = FirstLine("")

  /* Components of the request. */
  private var method: HttpMethod = _
  private var resource: String = _
  private var headers = Map[String, String]()
  private var data: String = _
  private var contentLength = -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"  => method = GET
      case "POST" => method = POST
      case _ => throw new Exception("Unknown HTTP method: " + parts(0))
    }
    resource = parts(1)
  }

  private def processHeaderLine(line :String) {
    line split ": " toList match {
      case List(key, value) =>
        headers += (key -> value)
        if (key == "Content-Length") contentLength = value.toInt
      case _ =>
    }
  }


  /**
   * 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)
      Request(conn, method, resource, headers, data)
    else
      throw new Exception("HTTP request not completed!")
  }
}