Introduce EncodeStream

core/src/main/scala/io/finch/EncodeStream.scala

@@ -0,0 +1,26 @@
+package io.finch
+
+import com.twitter.io.{Buf, Reader}
+import java.nio.charset.Charset
+
+/**
+ * A type-class that defines encoding of a stream in a shape of `S[F[_], A]` to Finagle's [[Reader]].
+ */
+trait EncodeStream[S[_[_], _], F[_], A] {
+
+  type ContentType <: String
+
+  def apply(s: S[F, A], cs: Charset): Reader[Buf]
+}
+
+object EncodeStream {
+
+  type Aux[S[_[_], _], F[_], A, CT <: String] =
+    EncodeStream[S, F, A] { type ContentType = CT }
+
+  type Json[S[_[_],_], F[_], A] = Aux[S, F, A, Application.Json]
+
+  type Text[S[_[_],_], F[_], A] = Aux[S, F, A, Text.Plain]
+}
+
+

core/src/main/scala/io/finch/Input.scala

@@ -3,7 +3,7 @@ package io.finch
 import cats.Eq
 import com.twitter.finagle.http.{Method, Request}
 import com.twitter.finagle.netty3.ChannelBufferBuf
-import com.twitter.io.Buf
+import com.twitter.io.{Buf, Reader}
 import java.nio.charset.{Charset, StandardCharsets}
 import org.jboss.netty.handler.codec.http.{DefaultHttpRequest, HttpMethod, HttpVersion}
 import org.jboss.netty.handler.codec.http.multipart.{DefaultHttpDataFactory, HttpPostRequestEncoder}
@@ -30,8 +30,21 @@ final case class Input(request: Request, route: List[String]) {
    * ```
    *  import io.finch._, io.circe._
    *
-   *  val text: Input = Input.post("/").withBody[Text.Plain]("Text Body")
-   *  val json: Input = Input.post("/").withBody[Application.Json](Map("json" -> "object"))
+   *  val text = Input.post("/").withBody[Text.Plain]("Text Body")
+   *  val json = Input.post("/").withBody[Application.Json](Map("json" -> "object"))
+   *```
+   *
+   * Also possible to create chunked inputs passing a stream as an argument.
+   *
+   *```
+   *  import io.finch._, io.finch.iteratee._, cats.effect.IO, io.iteratee.Enumerator
+   *  import io.finch.circe._, io.circe.generic.auto._
+   *
+   *  val enumerateText = Enumerator.enumerate[IO, String]("foo", "bar")
+   *  val text = Input.post("/").withBody[Text.Plain](enumerateText)
+   *
+   *  val enumerateJson = Enumerate.enumerate[IO, Map[String, String]](Map("foo" - "bar"))
+   *  val json = Input.post("/").withBody[Application.Json](enumerateJson)
    *```
    */
   def withBody[CT <: String]: Input.Body[CT] = new Input.Body[CT](this)
@@ -75,7 +88,7 @@ final case class Input(request: Request, route: List[String]) {
       )
     } else ChannelBufferBuf.Owned(req.getContent)
 
-    withBody[Application.WwwFormUrlencoded](content, Some(StandardCharsets.UTF_8))
+    withBody[Application.WwwFormUrlencoded](content, StandardCharsets.UTF_8)
   }
 }
 
@@ -84,12 +97,13 @@ final case class Input(request: Request, route: List[String]) {
  */
 object Input {
 
-  private final def copyRequest(from: Request): Request = {
-    val to = Request()
-    to.version = from.version
-    to.method = from.method
+  private final def copyRequest(from: Request): Request =
+    copyRequestWithReader(from, from.reader)
+
+  private final def copyRequestWithReader(from: Request, reader: Reader[Buf]): Request = {
+    val to = Request(from.version, from.method, from.uri, reader)
+    to.setChunked(from.isChunked)
     to.content = from.content
-    to.uri = from.uri
     from.headerMap.foreach { case (k, v) => to.headerMap.put(k, v) }
 
     to
@@ -99,17 +113,39 @@ object Input {
    * A helper class that captures the `Content-Type` of the payload.
    */
   class Body[CT <: String](i: Input) {
-    def apply[A](body: A, charset: Option[Charset] = None)(implicit
-      e: Encode.Aux[A, CT], w: Witness.Aux[CT]
-    ): Input = {
-      val content = e(body, charset.getOrElse(StandardCharsets.UTF_8))
+    def apply[A](body: A)(implicit e: Encode.Aux[A, CT], w: Witness.Aux[CT]): Input =
+      apply[A](body, StandardCharsets.UTF_8)
 
+    def apply[A](body: A, charset: Charset)(implicit
+      e: Encode.Aux[A, CT], W: Witness.Aux[CT]
+    ): Input = {
+      val content = e(body, charset)
       val copied = copyRequest(i.request)
 
+      copied.setChunked(false)
       copied.content = content
-      copied.contentType = w.value
+      copied.contentType = W.value
       copied.contentLength = content.length.toLong
-      charset.foreach(cs => copied.charset = cs.displayName().toLowerCase)
+      copied.charset = charset.displayName().toLowerCase
+
+      Input(copied, i.route)
+    }
+
+    def apply[S[_[_], _], F[_], A](s: S[F, A])(implicit
+      S: EncodeStream.Aux[S, F, A, CT], W: Witness.Aux[CT]
+    ): Input = apply[S, F, A](s, StandardCharsets.UTF_8)
+
+    def apply[S[_[_], _], F[_], A](s: S[F, A], charset: Charset)(implicit
+      S: EncodeStream.Aux[S, F, A, CT],
+      W: Witness.Aux[CT]
+    ): Input = {
+      val content = S(s, charset)
+      val copied = copyRequestWithReader(i.request, content)
+
+      copied.setChunked(true)
+      copied.contentType = W.value
+      copied.headerMap.setUnsafe("Transfer-Encoding", "chunked")
+      copied.charset = charset.displayName().toLowerCase
 
       Input(copied, i.route)
     }

core/src/main/scala/io/finch/ToResponse.scala

@@ -85,6 +85,17 @@ trait HighPriorityToResponseInstances extends LowPriorityToResponseInstances {
 
     rep
   }
+
+  implicit def streamToResponse[S[_[_], _], F[_], A, CT <: String](implicit
+    E: EncodeStream.Aux[S, F, A, CT],
+    W: Witness.Aux[CT]
+  ): Aux[S[F, A], CT] = instance { (a, cs) =>
+    val stream = E(a, cs)
+    val rep = Response(Version.Http11, Status.Ok, stream)
+    rep.headerMap.setUnsafe("Content-Type", W.value)
+
+    rep
+  }
 }
 
 object ToResponse extends HighPriorityToResponseInstances {

core/src/test/scala/io/finch/InputSpec.scala

@@ -3,7 +3,8 @@ package io.finch
 import java.nio.charset.Charset
 
 import com.twitter.finagle.http.Method
-import com.twitter.io.Buf
+import com.twitter.io.{Buf, Pipe, Reader}
+import com.twitter.util.{Await, Future}
 import io.finch.data.Foo
 import io.finch.internal.HttpContent
 
@@ -34,21 +35,42 @@ class InputSpec extends FinchSpec {
     }
   }
 
-  it should "add content through withBody" in {
+  it should "add fully-buffered content through withBody" in {
     check { (i: Input, b: Buf) =>
       i.withBody[Text.Plain](b).request.content === b
     }
   }
 
-  it should "add content corresponding to a class through withBody[JSON]" in {
-    implicit val encodeFoo: Encode.Json[Foo] = Encode.json(
-      (a, cs) => Buf.ByteArray.Owned(a.s.getBytes(cs.name))
-    )
+  it should "add chunked content throhg withBody" in {
+    type ListStream[F[_], A] = List[A]
+    implicit def listToReader[F[_], CT <: String]: EncodeStream.Aux[ListStream, F, Buf, CT] =
+      new EncodeStream[ListStream, F, Buf] {
+        type ContentType = CT
+
+        def apply(s: ListStream[F, Buf], cs: Charset): Reader[Buf] = {
+          val p = new Pipe[Buf]
+
+          def loop(from: List[Buf]): Future[Unit] = from match {
+            case h :: t => p.write(h).before(loop(t))
+            case _ => p.close()
+          }
+
+          loop(s)
+          p
+        }
+      }
 
+    check { (i: Input, s: List[Buf]) =>
+      val out = i.withBody[Application.OctetStream].apply[ListStream, Future, Buf](s).request.reader
+      s.forall(buf => buf == Await.result(out.read()).get)
+    }
+  }
+
+  it should "add content corresponding to a class through withBody[JSON]" in {
     check { (i: Input, f: Foo, cs: Charset) =>
-      val input = i.withBody[Application.Json](f, Some(cs))
+      val input = i.withBody[Application.Json](f, cs)
 
-      input.request.content.asString(cs) === f.s &&
+      input.request.content.asString(cs) === s"""{s:"${f.s}"""" &&
       input.request.contentType === Some(s"application/json;charset=${cs.displayName.toLowerCase}")
     }
   }

core/src/test/scala/io/finch/data/Foo.scala

@@ -29,6 +29,9 @@ object Foo {
       )
     )
 
+  implicit val encodeJsonFoo: Encode.Json[Foo] =
+    Encode.json((foo, cs) => Buf.ByteArray.Owned(s"""{s:"${foo.s}"""".getBytes(cs)))
+
   implicit val arbitraryFoo: Arbitrary[Foo] =
     Arbitrary(Gen.alphaStr.suchThat(_.nonEmpty).map(Foo.apply))
 }