Merge pull request #14 from rampart81/master

#3 Chunked Encoding 지원 추가

Author: eungjun-yi

build.sbt

@@ -0,0 +1,67 @@
+import sbt._
+import Keys._
+
+object ScalaHttpServerBuild extends Build {
+  import Dependencies._
+
+  lazy val root = Project(
+    id = "scala-http-server",
+    base = file("."),
+    settings = BuildSetting.settings ++ Seq(libraryDependencies ++= coreDeps)
+  )
+}
+
+object Resolvers {
+  val typesafe = "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"
+  val all      = Seq(typesafe)
+}
+
+object Dependencies {
+  val akka      = "com.typesafe.akka" % "akka-actor_2.10"   % "2.1.0"
+  val akkaTest  = "com.typesafe.akka" % "akka-testkit_2.10" % "2.1.0" % "test"
+  val tika      = "org.apache.tika"   % "tika-core"         % "1.2"
+  val scalaTest = "org.scalatest"     % "scalatest_2.10"    % "1.9.1" % "test"
+  val coreDeps  = Seq(
+    akka,
+    akkaTest,
+    tika,
+    scalaTest
+  )
+}
+
+object BuildSetting {
+  val org          = "laScala Korea"
+  val ver          = "1.0.0-SNAPSHOT"
+  val scalaVer     = "2.10.0"
+  val settings     = Defaults.defaultSettings ++ Seq(
+        organization := org,
+        version      := ver,
+        scalaVersion := scalaVer,
+        shellPrompt  := SBTPrompt.setting,
+        resolvers    := Resolvers.all)
+}
+
+// SBT Prompt 를 튜닝하기 위한 세팅.
+object SBTPrompt {
+
+  object devnull extends ProcessLogger {
+    def info(s: => String) {}
+    def error(s: => String) {}
+    def buffer[T](f: => T): T = f
+  }
+
+  def gitBranches = ("git branch --no-color" lines_! devnull mkString)
+
+  val current = """\*\s+([\w-/]+)""".r
+  val setting = {
+    (state: State) => {
+      val currBranch =
+        current findFirstMatchIn gitBranches map (_ group (1)) getOrElse "-"
+      val currProject = Project.extract(state).currentProject.id
+      "%s:%s:%s> ".format(
+        currProject, currBranch, BuildSetting.ver
+      )
+    }
+  }
+}
+

project/Build.scala

@@ -19,14 +19,17 @@ package com.lascala.http
 
 import com.lascala.http.HttpConstants._
 
-import akka.util.{ ByteString, ByteStringBuilder }
+import akka.util.{ByteString, ByteStringBuilder}
 import java.io.File
 import org.apache.tika.Tika
 import java.io.FileInputStream
 import java.util.Date
 import java.util.Locale
 import java.text.SimpleDateFormat
 import java.util.TimeZone
+import akka.actor.IO
+import util.Failure
+import com.lascala.libs.Enumerator
 
 trait HttpResponse {
   def lastModified: Date = null
@@ -42,12 +45,12 @@ trait HttpResponse {
 }
 
 object HttpResponse {
-  val version = ByteString("HTTP/1.1")
-  val server = ByteString("Server: lascala-http")
-  val connection = ByteString("Connection: ")
-  val keepAlive = ByteString("Keep-Alive")
-  val close = ByteString("Close")
-  val date = ByteString("Date: ")
+  val version      = ByteString("HTTP/1.1")
+  val server       = ByteString("Server: lascala-http")
+  val connection   = ByteString("Connection: ")
+  val keepAlive    = ByteString("Keep-Alive")
+  val close        = ByteString("Close")
+  val date         = ByteString("Date: ")
   val lastModified = ByteString("Last-Modified: ")
 
   def httpDateFormat = {
@@ -69,26 +72,71 @@ object HttpResponse {
     rsp.contentLength ++= CRLF ++=
     connection ++= (if (rsp.shouldKeepAlive) keepAlive else close) ++= CRLF ++= CRLF ++= rsp.body).result
   }
+
+  def stream(rsp: HttpResponse with ChunkedEncodable, socket: IO.SocketHandle) = {
+    val headers = (new ByteStringBuilder ++=
+      version ++= SP ++= rsp.status ++= SP ++= rsp.reason ++= CRLF ++=
+      rsp.contentType ++ CRLF ++=
+      rsp.cacheControl ++= CRLF ++=
+      date ++= httpDate(new Date) ++= CRLF ++=
+      Option(rsp.lastModified).map(lastModified ++ httpDate(_) ++ CRLF).getOrElse(ByteString("")) ++=
+      server ++= CRLF ++=
+      connection ++= (if (rsp.shouldKeepAlive) keepAlive else close) ++= CRLF ++=
+      ByteString("Transfer-Encoding: chunked") ++= CRLF ++= CRLF).result
+
+    socket write headers.compact
+
+    rsp.chunkedData.foreach { chunk =>
+      val chunkedMessageBody = (
+        new ByteStringBuilder ++= ByteString(chunk.size.toHexString) ++= CRLF
+          ++= chunk ++= CRLF).result
+
+      socket write chunkedMessageBody.compact
+    }
+
+    // According to the HTTP spec, need to write 0 at the end of the chunk data
+    socket write ByteString("0") ++ CRLF ++ CRLF
+  }
+}
+
+trait ChunkedEncodable extends HttpResponse {
+  def chunkedData: Enumerator[ByteString]
 }
 
 case class OKFileResponse(file: File, shouldKeepAlive: Boolean = true) extends HttpResponse {
+  val body     = readFile(file)
+  val mimeType = new Tika().detect(file)
+  val status   = ByteString("200")
+  val reason   = ByteString("OK")
+ 
+ override def lastModified = new Date(file.lastModified)
 	def readFile(file: File) = {
     val resource = new Array[Byte](file.length.toInt)
     val in = new FileInputStream(file)
     in.read(resource)
     in.close()
     ByteString(resource)
   }
-  val body = readFile(file)
-  val mimeType = new Tika().detect(file)
-  val status = ByteString("200")
-  val reason = ByteString("OK")
-  override def lastModified = new Date(file.lastModified)
 }
 
 case class OKResponse(body: ByteString, shouldKeepAlive: Boolean = true, mimeType: String = "text/html") extends HttpResponse {
   val status = ByteString("200")
   val reason = ByteString("OK")
+
+  def withMimeType(mimeType: String) = this match {
+    // In case of ChunkedEncodable, need to manually instantiate a new OKResponse with ChunkedEncodable 
+    // Instead of just using copy method in order to preserve the ChunkedEncodable type.
+    case t: ChunkedEncodable => new OKResponse(this.body, this.shouldKeepAlive, mimeType) with ChunkedEncodable {
+      def chunkedData: Enumerator[ByteString] = t.chunkedData
+    }
+    case _ => this.copy(mimeType = mimeType)
+  }
+}
+
+object OKResponse {
+  def stream(chunk: Enumerator[ByteString]) = new OKResponse(body = ByteString.empty, mimeType = "text/html") with ChunkedEncodable {
+    def chunkedData: Enumerator[ByteString] = chunk
+  }
 }
 
 case class NotModifiedResponse(body: ByteString = ByteString.empty, shouldKeepAlive: Boolean = false, mimeType: String = "") extends HttpResponse {

src/main/resource/stream_data1.txt

@@ -29,7 +29,12 @@ object HttpServer {
 
   def processRequest(socket: IO.SocketHandle, handler: ActorRef): IO.Iteratee[Unit] = {
     def processResponse(response: HttpResponse) {
-      socket write HttpResponse.bytes(response).compact
+      response match {
+        case chunkEncoded: HttpResponse with ChunkedEncodable =>
+          HttpResponse.stream(chunkEncoded, socket)
+        case _ => socket write HttpResponse.bytes(response).compact
+      }
+      
       if (!response.shouldKeepAlive) socket.close()
     }
 

src/main/resource/stream_data2.txt

@@ -0,0 +1,132 @@
+/*
+ * This software is licensed under the Apache 2 license, quoted below.
+ *  
+ *  Copyright 2009-2012 Typesafe Inc. <http://www.typesafe.com>
+ *  
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *  
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License. 
+ */
+package com.lascala.libs
+
+import akka.actor.IO
+import akka.util.ByteString
+import concurrent.{Await, Future}
+import concurrent.ExecutionContext.Implicits.global
+import java.io.InputStream
+import util.{Success, Failure}
+import concurrent.duration.Duration
+import java.util.concurrent.TimeUnit
+
+
+/**
+ * Chunked encoding 을 서포트 하기위해 기본적인 Enumerator(or Producer) 패턴 을 구현한것이다.
+ * future 가 각각 처리되어야 할 데이타 를 담당한다.
+ * Foreach 메소드 가 이 Enumerator 와 연결되어있는 다른 enumerator 를 연동해
+ * 각 Enumerator 의 future 를 호출해서 각각 chunk 데이타를 처리한다. 
+ *
+ */
+trait Enumerator[T] {
+  parent =>
+
+  /**
+   * 스칼라 Future 를 사용해서 인풋을 처리함.
+   * 
+   * @note Some 은 계속해서 처리할 인풋이 있다는 뜻이고 None 은 그 반대임.
+   */
+  def future: Future[Option[T]]
+
+ /**
+  * 각각 Enumerator 의 퓨처를 호출한후 리턴된 값을 주어진 콜백 function 에 
+  * 적용한다. 퓨처는 None 을 리턴하기까지 귀재 방식으로 계속해서 호출됨.
+  *
+  * @param f callbakc function 을 사용하요 퓨처가 리턴한 값을 처리함. 
+  */
+ @scala.annotation.tailrec
+  final def foreach[E](f: T => E) {
+    Await.result(future, Duration.apply(60, TimeUnit.SECONDS)) match {
+      case Some(v) => {
+        f(v)
+        foreach(f)
+      }
+      case None => if (next.isDefined) next.get.foreach(f)
+    }
+  }
+
+  def next: Option[Enumerator[T]] = None
+
+  /**
+   * 여러 Enumerator 들을 연결시켜 주는 메소드
+   *
+   * @param other
+   * @return
+   */
+  def andThen(other: Enumerator[T]) = {
+    def _chainOtherEnumerators(current: Option[Enumerator[T]], other: Enumerator[T]): Option[Enumerator[T]] = current match {
+      case Some(curr) => Some(new Enumerator[T] {
+        def future = curr.future
+        override def next = _chainOtherEnumerators(curr.next, other)
+      })
+      case None => Some(other)
+    }
+
+    new Enumerator[T] {
+      def future: Future[Option[T]] = parent.future
+      override def next = _chainOtherEnumerators(parent.next, other)
+    }
+  }
+}
+
+/**
+ * Enumerator companion object.
+ */
+object Enumerator {
+  
+  def fromCallback(callback: () => Option[ByteString]) = new Enumerator[ByteString] {
+    def future = Future(callback())
+  }
+
+  /**
+   * Takes InputStream as input source and returns a Enumerator that produces data.
+   * InputStream 을 받아서 Enumerator 를 리턴해주는 유틸펑션.
+   *
+   * @param in
+   * @param chunkSize
+   */
+  def fromStream(in: InputStream, chunkSize: Int = 1024 * 8) = {
+    def callback = { () =>
+      val buffer = new Array[Byte](chunkSize)
+      val chunk = in.read(buffer) match {
+        case -1 =>
+          in.close()
+          None
+        case read =>
+          val input = new Array[Byte](read)
+          System.arraycopy(buffer, 0, input, 0, read)
+          Some(ByteString(input))
+      }
+
+      chunk
+    }
+
+    fromCallback(callback)
+  }
+
+  /**
+   * Convinient function that takes a file and returns a Enumerator
+   * 
+   * 파일 오브젝트를 받아서 Enumerator 를 리턴해주는 유틸펑션.
+   * 
+   * @param file
+   * @param chunkSize
+   */
+  def fromFile(file: java.io.File, chunkSize: Int = 1024 * 8) = fromStream(new java.io.FileInputStream(file), chunkSize)
+}

src/main/scala/com/lascala/http/HttpResponse.scala

@@ -29,6 +29,8 @@ import com.lascala.http.HttpResponse._
 import com.lascala.http.HttpConstants._
 import akka.util.ByteString
 import java.io.File
+import com.lascala.libs.Enumerator
+import org.apache.tika.Tika
 
 /**
  * Sample Demo Application
@@ -43,16 +45,30 @@ class RequestHandler extends Actor {
   val docroot = "."
 
   def receive = {
-    case HttpRequest("GET", pathSegments, _, _, headers, _) =>
+    case HttpRequest("GET", List("chunked"), _, _, headers, _) => 
+      sender !  OKResponse.stream(getEnumerator)
+    case HttpRequest("GET", List("download"), _, _, headers, _) =>
+      sender !  OKResponse.stream(getEnumerator).withMimeType("text/plain")
+    case HttpRequest("GET", pathSegments, _, _, headers, _) => 
       new File(docroot, "/" + pathSegments.mkString(File.separator)) match {
         case file if file.isFile() =>
           headers.find(_.name.toLowerCase == "if-modified-since") match {
             case Some(d) if HttpResponse.httpDateFormat.parse(d.value).compareTo(new java.util.Date(file.lastModified)) != -1 => sender ! NotModifiedResponse()
             case _ => sender ! OKFileResponse(file, true)
           }
-        case _ =>
-          sender ! NotFoundError()
+        case _ => sender ! NotFoundError()
       }
     case _ => sender ! MethodNotAllowedError()
   }
+
+  private def getEnumerator = {
+    // chunked encoding 을 사용하여 사이즈가 큰 파일을 스트리밍으로 처리하기 받기.
+    val file1 = new File(docroot + "/src/main/resource", "stream_data1.txt")
+    val file2 = new File(docroot + "/src/main/resource", "stream_data2.txt")
+    val enum1 = Enumerator.fromFile(file1)
+    val enum2 = Enumerator.fromFile(file2) 
+
+    // enume 을 연결시켜서 여러가지 input source 를 한 스트림으로 순서대로 처리하기.
+    enum1 andThen enum2  
+  }
 }