From d65fe4e04ecb42c50d8c2fe0a0081399b938a631 Mon Sep 17 00:00:00 2001
From: Mathias <mathias@decodified.com>
Date: Wed, 12 Nov 2014 14:27:30 +0100
Subject: [PATCH] !htp #16190 refactor and improve marshalling infrastructure,
 get rid of `Marshallers` abstraction altogether

Also improve Unmarshalling infrastructure once more.
---
 .../http/testkit/MarshallingTestUtils.scala   |   2 +-
 .../akka/http/client/RequestBuilding.scala    |   8 +-
 .../http/marshalling/GenericMarshallers.scala |   2 +-
 .../scala/akka/http/marshalling/Marshal.scala |  27 ++--
 .../akka/http/marshalling/Marshaller.scala    | 134 ++++++++++++++++++
 .../akka/http/marshalling/Marshallers.scala   | 111 ---------------
 .../marshalling/MultipartMarshallers.scala    |  28 ++--
 .../PredefinedToEntityMarshallers.scala       |   5 +-
 .../PredefinedToRequestMarshallers.scala      |   6 +-
 .../PredefinedToResponseMarshallers.scala     |   2 +-
 .../marshalling/ToResponseMarshallable.scala  |   8 +-
 .../scala/akka/http/marshalling/package.scala |   4 -
 .../unmarshalling/GenericUnmarshallers.scala  |  11 +-
 .../MultipartUnmarshallers.scala              |   4 +-
 .../PredefinedFromEntityUnmarshallers.scala   |   9 +-
 .../PredefinedFromStringUnmarshallers.scala   |   5 +-
 .../http/unmarshalling/Unmarshaller.scala     |  82 +++++++----
 17 files changed, 250 insertions(+), 198 deletions(-)
 create mode 100644 akka-http/src/main/scala/akka/http/marshalling/Marshaller.scala
 delete mode 100644 akka-http/src/main/scala/akka/http/marshalling/Marshallers.scala

diff --git a/akka-http-testkit/src/main/scala/akka/http/testkit/MarshallingTestUtils.scala b/akka-http-testkit/src/main/scala/akka/http/testkit/MarshallingTestUtils.scala
index e2c7589d1d..6e33777cb4 100644
--- a/akka-http-testkit/src/main/scala/akka/http/testkit/MarshallingTestUtils.scala
+++ b/akka-http-testkit/src/main/scala/akka/http/testkit/MarshallingTestUtils.scala
@@ -16,7 +16,7 @@ import akka.stream.FlowMaterializer
 import scala.util.Try
 
 trait MarshallingTestUtils {
-  def marshal[T: ToEntityMarshallers](value: T)(implicit ec: ExecutionContext, mat: FlowMaterializer): HttpEntity.Strict =
+  def marshal[T: ToEntityMarshaller](value: T)(implicit ec: ExecutionContext, mat: FlowMaterializer): HttpEntity.Strict =
     Await.result(Marshal(value).to[HttpEntity].flatMap(_.toStrict(1.second)), 1.second)
 
   def unmarshalValue[T: FromEntityUnmarshaller](entity: HttpEntity)(implicit ec: ExecutionContext, mat: FlowMaterializer): T =
diff --git a/akka-http/src/main/scala/akka/http/client/RequestBuilding.scala b/akka-http/src/main/scala/akka/http/client/RequestBuilding.scala
index 28a8fd4307..6e9f81fa48 100644
--- a/akka-http/src/main/scala/akka/http/client/RequestBuilding.scala
+++ b/akka-http/src/main/scala/akka/http/client/RequestBuilding.scala
@@ -26,10 +26,10 @@ trait RequestBuilding extends TransformerPipelineSupport {
     def apply(uri: String): HttpRequest =
       apply(uri, HttpEntity.Empty)
 
-    def apply[T](uri: String, content: T)(implicit m: ToEntityMarshallers[T], ec: ExecutionContext): HttpRequest =
+    def apply[T](uri: String, content: T)(implicit m: ToEntityMarshaller[T], ec: ExecutionContext): HttpRequest =
       apply(uri, Some(content))
 
-    def apply[T](uri: String, content: Option[T])(implicit m: ToEntityMarshallers[T], ec: ExecutionContext): HttpRequest =
+    def apply[T](uri: String, content: Option[T])(implicit m: ToEntityMarshaller[T], ec: ExecutionContext): HttpRequest =
       apply(Uri(uri), content)
 
     def apply(uri: String, entity: RequestEntity): HttpRequest =
@@ -38,10 +38,10 @@ trait RequestBuilding extends TransformerPipelineSupport {
     def apply(uri: Uri): HttpRequest =
       apply(uri, HttpEntity.Empty)
 
-    def apply[T](uri: Uri, content: T)(implicit m: ToEntityMarshallers[T], ec: ExecutionContext): HttpRequest =
+    def apply[T](uri: Uri, content: T)(implicit m: ToEntityMarshaller[T], ec: ExecutionContext): HttpRequest =
       apply(uri, Some(content))
 
-    def apply[T](uri: Uri, content: Option[T])(implicit m: ToEntityMarshallers[T], timeout: Timeout = Timeout(1.second), ec: ExecutionContext): HttpRequest =
+    def apply[T](uri: Uri, content: Option[T])(implicit m: ToEntityMarshaller[T], timeout: Timeout = Timeout(1.second), ec: ExecutionContext): HttpRequest =
       content match {
         case None ⇒ apply(uri, HttpEntity.Empty)
         case Some(value) ⇒
diff --git a/akka-http/src/main/scala/akka/http/marshalling/GenericMarshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/GenericMarshallers.scala
index c390a9ce7e..2f7043f50f 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/GenericMarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/GenericMarshallers.scala
@@ -16,7 +16,7 @@ trait GenericMarshallers extends LowPriorityToResponseMarshallerImplicits {
   implicit def optionMarshaller[A, B](implicit m: Marshaller[A, B], empty: EmptyValue[B]): Marshaller[Option[A], B] =
     Marshaller {
       case Some(value) ⇒ m(value)
-      case None        ⇒ FastFuture.successful(Marshalling.Opaque(() ⇒ empty.emptyValue))
+      case None        ⇒ FastFuture.successful(Marshalling.Opaque(() ⇒ empty.emptyValue) :: Nil)
     }
 
   implicit def eitherMarshaller[A1, A2, B](implicit m1: Marshaller[A1, B], m2: Marshaller[A2, B]): Marshaller[Either[A1, A2], B] =
diff --git a/akka-http/src/main/scala/akka/http/marshalling/Marshal.scala b/akka-http/src/main/scala/akka/http/marshalling/Marshal.scala
index 456f04f1e9..d935230f54 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/Marshal.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/Marshal.scala
@@ -4,7 +4,6 @@
 
 package akka.http.marshalling
 
-import scala.collection.immutable
 import scala.concurrent.{ ExecutionContext, Future }
 import akka.http.util.FastFuture
 import akka.http.model.HttpCharsets._
@@ -14,36 +13,36 @@ import FastFuture._
 object Marshal {
   def apply[T](value: T): Marshal[T] = new Marshal(value)
 
-  class UnacceptableResponseContentTypeException(supported: immutable.Seq[ContentType]) extends RuntimeException
+  case class UnacceptableResponseContentTypeException(supported: Set[ContentType]) extends RuntimeException
 
   private class MarshallingWeight(val weight: Float, val marshal: () ⇒ HttpResponse)
 }
 
 class Marshal[A](val value: A) {
   /**
-   * Marshals `value` to the first of the available `Marshallers` for `A` and `B`.
-   * If the marshaller is flexible with regard to the used charset `UTF-8` is chosen.
+   * Marshals `value` using the first available [[Marshalling]] for `A` and `B` provided by the given [[Marshaller]].
+   * If the marshalling is flexible with regard to the used charset `UTF-8` is chosen.
    */
-  def to[B](implicit m: Marshallers[A, B], ec: ExecutionContext): Future[B] =
-    m.marshallers.head(value).fast.map {
-      case Marshalling.WithFixedCharset(_, _, marshal) ⇒ marshal()
-      case Marshalling.WithOpenCharset(_, marshal)     ⇒ marshal(HttpCharsets.`UTF-8`)
-      case Marshalling.Opaque(marshal)                 ⇒ marshal()
+  def to[B](implicit m: Marshaller[A, B], ec: ExecutionContext): Future[B] =
+    m(value).fast.map {
+      _.head match {
+        case Marshalling.WithFixedCharset(_, _, marshal) ⇒ marshal()
+        case Marshalling.WithOpenCharset(_, marshal)     ⇒ marshal(HttpCharsets.`UTF-8`)
+        case Marshalling.Opaque(marshal)                 ⇒ marshal()
+      }
     }
 
   /**
    * Marshals `value` to an `HttpResponse` for the given `HttpRequest` with full content-negotiation.
    */
-  def toResponseFor(request: HttpRequest)(implicit m: ToResponseMarshallers[A], ec: ExecutionContext): Future[HttpResponse] = {
+  def toResponseFor(request: HttpRequest)(implicit m: ToResponseMarshaller[A], ec: ExecutionContext): Future[HttpResponse] = {
     import akka.http.marshalling.Marshal._
     val mediaRanges = request.acceptedMediaRanges // cache for performance
     val charsetRanges = request.acceptedCharsetRanges // cache for performance
     def qValueMT(mediaType: MediaType) = request.qValueForMediaType(mediaType, mediaRanges)
     def qValueCS(charset: HttpCharset) = request.qValueForCharset(charset, charsetRanges)
 
-    val marshallingFutures = m.marshallers.map(_(value))
-    val marshallingsFuture = FastFuture.sequence(marshallingFutures)
-    marshallingsFuture.fast.map { marshallings ⇒
+    m(value).fast.map { marshallings ⇒
       def weight(mt: MediaType, cs: HttpCharset): Float = math.min(qValueMT(mt), qValueCS(cs))
       val defaultMarshallingWeight: MarshallingWeight =
         new MarshallingWeight(0f, { () ⇒
@@ -51,7 +50,7 @@ class Marshal[A](val value: A) {
             case Marshalling.WithFixedCharset(mt, cs, _) ⇒ ContentType(mt, cs)
             case Marshalling.WithOpenCharset(mt, _)      ⇒ ContentType(mt)
           }
-          throw new UnacceptableResponseContentTypeException(supportedContentTypes)
+          throw UnacceptableResponseContentTypeException(supportedContentTypes.toSet)
         })
       val best = marshallings.foldLeft(defaultMarshallingWeight) {
         case (acc, Marshalling.WithFixedCharset(mt, cs, marshal)) ⇒
diff --git a/akka-http/src/main/scala/akka/http/marshalling/Marshaller.scala b/akka-http/src/main/scala/akka/http/marshalling/Marshaller.scala
new file mode 100644
index 0000000000..db8594f7f7
--- /dev/null
+++ b/akka-http/src/main/scala/akka/http/marshalling/Marshaller.scala
@@ -0,0 +1,134 @@
+/**
+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.http.marshalling
+
+import scala.concurrent.{ Future, ExecutionContext }
+import scala.util.control.NonFatal
+import akka.http.util.FastFuture
+import akka.http.model._
+import FastFuture._
+
+sealed abstract class Marshaller[-A, +B] extends (A ⇒ Future[List[Marshalling[B]]]) {
+
+  def map[C](f: B ⇒ C)(implicit ec: ExecutionContext): Marshaller[A, C] =
+    Marshaller[A, C](value ⇒ this(value).fast map (_ map (_ map f)))
+
+  /**
+   * Reuses this Marshaller's logic to produce a new Marshaller from another type `C` which overrides
+   * the produced [[ContentType]] with another one.
+   * Depending on whether the given [[ContentType]] has a defined charset or not and whether the underlying
+   * marshaller marshals with a fixed charset it can happen, that the wrapping becomes illegal.
+   * For example, a marshaller producing content encoded with UTF-16 cannot be wrapped with a [[ContentType]]
+   * that has a defined charset of UTF-8, since akka-http will never recode entities.
+   * If the wrapping is illegal the [[Future]] produced by the resulting marshaller will contain a [[RuntimeException]].
+   */
+  def wrap[C, D >: B](contentType: ContentType)(f: C ⇒ A)(implicit ec: ExecutionContext, mto: MediaTypeOverrider[D]): Marshaller[C, D] =
+    Marshaller { value ⇒
+      import Marshalling._
+      this(f(value)).fast map {
+        _ map {
+          case WithFixedCharset(_, cs, marshal) if contentType.definedCharset.isEmpty || contentType.charset == cs ⇒
+            WithFixedCharset(contentType.mediaType, cs, () ⇒ mto(marshal(), contentType.mediaType))
+          case WithOpenCharset(_, marshal) if contentType.definedCharset.isEmpty ⇒
+            WithOpenCharset(contentType.mediaType, cs ⇒ mto(marshal(cs), contentType.mediaType))
+          case WithOpenCharset(_, marshal) ⇒
+            WithFixedCharset(contentType.mediaType, contentType.charset, () ⇒ mto(marshal(contentType.charset), contentType.mediaType))
+          case Opaque(marshal) if contentType.definedCharset.isEmpty ⇒ Opaque(() ⇒ mto(marshal(), contentType.mediaType))
+          case x ⇒ sys.error(s"Illegal marshaller wrapping. Marshalling `$x` cannot be wrapped with ContentType `$contentType`")
+        }
+      }
+    }
+
+  override def compose[C](f: C ⇒ A): Marshaller[C, B] = Marshaller(super.compose(f))
+}
+
+object Marshaller
+  extends GenericMarshallers
+  with PredefinedToEntityMarshallers
+  with PredefinedToResponseMarshallers
+  with PredefinedToRequestMarshallers {
+
+  /**
+   * Creates a [[Marshaller]] from the given function.
+   */
+  def apply[A, B](f: A ⇒ Future[List[Marshalling[B]]]): Marshaller[A, B] =
+    new Marshaller[A, B] {
+      def apply(value: A) =
+        try f(value)
+        catch { case NonFatal(e) ⇒ FastFuture.failed(e) }
+    }
+
+  /**
+   * Helper for creating a [[Marshaller]] using the given function.
+   */
+  def strict[A, B](f: A ⇒ Marshalling[B]): Marshaller[A, B] =
+    Marshaller { a ⇒ FastFuture.successful(f(a) :: Nil) }
+
+  /**
+   * Helper for creating a "super-marshaller" from a number of "sub-marshallers".
+   * Content-negotiation determines, which "sub-marshallers" eventually gets to do the job.
+   */
+  def oneOf[A, B](marshallers: Marshaller[A, B]*)(implicit ec: ExecutionContext): Marshaller[A, B] =
+    Marshaller { a ⇒ FastFuture.sequence(marshallers.map(_(a))).fast.map(_.flatten.toList) }
+
+  /**
+   * Helper for creating a "super-marshaller" from a number of values and a function producing "sub-marshallers"
+   * from these values. Content-negotiation determines, which "sub-marshallers" eventually gets to do the job.
+   */
+  def oneOf[T, A, B](values: T*)(f: T ⇒ Marshaller[A, B])(implicit ec: ExecutionContext): Marshaller[A, B] =
+    oneOf(values map f: _*)
+
+  /**
+   * Helper for creating a synchronous [[Marshaller]] to content with a fixed charset from the given function.
+   */
+  def withFixedCharset[A, B](mediaType: MediaType, charset: HttpCharset)(marshal: A ⇒ B): Marshaller[A, B] =
+    strict { value ⇒ Marshalling.WithFixedCharset(mediaType, charset, () ⇒ marshal(value)) }
+
+  /**
+   * Helper for creating a synchronous [[Marshaller]] to content with a negotiable charset from the given function.
+   */
+  def withOpenCharset[A, B](mediaType: MediaType)(marshal: (A, HttpCharset) ⇒ B): Marshaller[A, B] =
+    strict { value ⇒ Marshalling.WithOpenCharset(mediaType, charset ⇒ marshal(value, charset)) }
+
+  /**
+   * Helper for creating a synchronous [[Marshaller]] to non-negotiable content from the given function.
+   */
+  def opaque[A, B](marshal: A ⇒ B): Marshaller[A, B] =
+    strict { value ⇒ Marshalling.Opaque(() ⇒ marshal(value)) }
+}
+
+/**
+ * Describes one possible option for marshalling a given value.
+ */
+sealed trait Marshalling[+A] {
+  def map[B](f: A ⇒ B): Marshalling[B]
+}
+
+object Marshalling {
+  /**
+   * A Marshalling to a specific MediaType and charset.
+   */
+  final case class WithFixedCharset[A](mediaType: MediaType,
+                                       charset: HttpCharset,
+                                       marshal: () ⇒ A) extends Marshalling[A] {
+    def map[B](f: A ⇒ B): WithFixedCharset[B] = copy(marshal = () ⇒ f(marshal()))
+  }
+
+  /**
+   * A Marshalling to a specific MediaType and a potentially flexible charset.
+   */
+  final case class WithOpenCharset[A](mediaType: MediaType,
+                                      marshal: HttpCharset ⇒ A) extends Marshalling[A] {
+    def map[B](f: A ⇒ B): WithOpenCharset[B] = copy(marshal = cs ⇒ f(marshal(cs)))
+  }
+
+  /**
+   * A Marshalling to an unknown MediaType and charset.
+   * Circumvents content negotiation.
+   */
+  final case class Opaque[A](marshal: () ⇒ A) extends Marshalling[A] {
+    def map[B](f: A ⇒ B): Opaque[B] = copy(marshal = () ⇒ f(marshal()))
+  }
+}
diff --git a/akka-http/src/main/scala/akka/http/marshalling/Marshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/Marshallers.scala
deleted file mode 100644
index 9b1445e151..0000000000
--- a/akka-http/src/main/scala/akka/http/marshalling/Marshallers.scala
+++ /dev/null
@@ -1,111 +0,0 @@
-/**
- * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
- */
-
-package akka.http.marshalling
-
-import scala.collection.immutable
-import scala.concurrent.{ Future, ExecutionContext }
-import scala.util.control.NonFatal
-import akka.http.util.FastFuture
-import akka.http.model._
-import FastFuture._
-
-case class Marshallers[-A, +B](marshallers: immutable.Seq[Marshaller[A, B]]) {
-  require(marshallers.nonEmpty, "marshallers must be non-empty")
-  def map[C](f: B ⇒ C)(implicit ec: ExecutionContext): Marshallers[A, C] =
-    Marshallers(marshallers map (_ map f))
-}
-
-object Marshallers extends SingleMarshallerMarshallers {
-  def apply[A, B](m: Marshaller[A, B]): Marshallers[A, B] = apply(m :: Nil)
-  def apply[A, B](first: Marshaller[A, B], more: Marshaller[A, B]*): Marshallers[A, B] = apply(first +: more.toVector)
-  def apply[A, B](mediaTypes: MediaType*)(f: MediaType ⇒ Marshaller[A, B]): Marshallers[A, B] =
-    Marshallers(mediaTypes.map(f)(collection.breakOut))
-
-  implicit def entity2response[T](implicit m: Marshallers[T, ResponseEntity], ec: ExecutionContext): Marshallers[T, HttpResponse] =
-    m map (entity ⇒ HttpResponse(entity = entity))
-}
-
-sealed abstract class SingleMarshallerMarshallers {
-  implicit def singleMarshallerMarshallers[A, B](implicit m: Marshaller[A, B]): Marshallers[A, B] = Marshallers(m)
-}
-
-sealed trait Marshaller[-A, +B] extends (A ⇒ Future[Marshalling[B]]) {
-
-  def map[C](f: B ⇒ C)(implicit ec: ExecutionContext): Marshaller[A, C] =
-    Marshaller[A, C](value ⇒ this(value).fast.map(_ map f))
-
-  /**
-   * Reuses this Marshaller's logic to produce a new Marshaller from another type `C` which overrides
-   * the produced media-type with another one.
-   */
-  def wrap[C, D >: B](mediaType: MediaType)(f: C ⇒ A)(implicit ec: ExecutionContext, mto: MediaTypeOverrider[D]): Marshaller[C, D] =
-    Marshaller { value ⇒
-      import Marshalling._
-      this(f(value)).fast.map {
-        case WithFixedCharset(_, cs, marshal) ⇒ WithFixedCharset(mediaType, cs, () ⇒ mto(marshal(), mediaType))
-        case WithOpenCharset(_, marshal)      ⇒ WithOpenCharset(mediaType, cs ⇒ mto(marshal(cs), mediaType))
-        case Opaque(marshal)                  ⇒ Opaque(() ⇒ mto(marshal(), mediaType))
-      }
-    }
-
-  override def compose[C](f: C ⇒ A): Marshaller[C, B] = Marshaller(super.compose(f))
-}
-
-object Marshaller
-  extends GenericMarshallers
-  with PredefinedToEntityMarshallers
-  with PredefinedToResponseMarshallers
-  with PredefinedToRequestMarshallers {
-
-  def apply[A, B](f: A ⇒ Future[Marshalling[B]]): Marshaller[A, B] =
-    new Marshaller[A, B] {
-      def apply(value: A) =
-        try f(value)
-        catch { case NonFatal(e) ⇒ FastFuture.failed(e) }
-    }
-
-  def withFixedCharset[A, B](mediaType: MediaType, charset: HttpCharset)(marshal: A ⇒ B): Marshaller[A, B] =
-    Marshaller { value ⇒ FastFuture.successful(Marshalling.WithFixedCharset(mediaType, charset, () ⇒ marshal(value))) }
-
-  def withOpenCharset[A, B](mediaType: MediaType)(marshal: (A, HttpCharset) ⇒ B): Marshaller[A, B] =
-    Marshaller { value ⇒ FastFuture.successful(Marshalling.WithOpenCharset(mediaType, charset ⇒ marshal(value, charset))) }
-
-  def opaque[A, B](marshal: A ⇒ B): Marshaller[A, B] =
-    Marshaller { value ⇒ FastFuture.successful(Marshalling.Opaque(() ⇒ marshal(value))) }
-}
-
-/**
- * Describes what a Marshaller can produce for a given value.
- */
-sealed trait Marshalling[+A] {
-  def map[B](f: A ⇒ B): Marshalling[B]
-}
-
-object Marshalling {
-  /**
-   * A Marshalling to a specific MediaType and charset.
-   */
-  final case class WithFixedCharset[A](mediaType: MediaType,
-                                       charset: HttpCharset,
-                                       marshal: () ⇒ A) extends Marshalling[A] {
-    def map[B](f: A ⇒ B): WithFixedCharset[B] = copy(marshal = () ⇒ f(marshal()))
-  }
-
-  /**
-   * A Marshalling to a specific MediaType and a potentially flexible charset.
-   */
-  final case class WithOpenCharset[A](mediaType: MediaType,
-                                      marshal: HttpCharset ⇒ A) extends Marshalling[A] {
-    def map[B](f: A ⇒ B): WithOpenCharset[B] = copy(marshal = cs ⇒ f(marshal(cs)))
-  }
-
-  /**
-   * A Marshalling to an unknown MediaType and charset.
-   * Circumvents content negotiation.
-   */
-  final case class Opaque[A](marshal: () ⇒ A) extends Marshalling[A] {
-    def map[B](f: A ⇒ B): Opaque[B] = copy(marshal = () ⇒ f(marshal()))
-  }
-}
diff --git a/akka-http/src/main/scala/akka/http/marshalling/MultipartMarshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/MultipartMarshallers.scala
index 15dc65e10f..698b34f9ca 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/MultipartMarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/MultipartMarshallers.scala
@@ -27,23 +27,21 @@ trait MultipartMarshallers {
   }
 
   implicit def multipartMarshaller[T <: Multipart](implicit log: LoggingAdapter = NoLogging): ToEntityMarshaller[T] =
-    Marshaller { value ⇒
+    Marshaller strict { value ⇒
       val boundary = randomBoundary
       val contentType = ContentType(value.mediaType withBoundary boundary)
-      FastFuture.successful {
-        Marshalling.WithOpenCharset(contentType.mediaType, { charset ⇒
-          value match {
-            case x: Multipart.Strict ⇒
-              val data = BodyPartRenderer.strict(x.strictParts, boundary, charset.nioCharset, partHeadersSizeHint = 128, log)
-              HttpEntity(contentType, data)
-            case _ ⇒
-              val chunks = value.parts
-                .transform("bodyPartRenderer", () ⇒ BodyPartRenderer.streamed(boundary, charset.nioCharset, partHeadersSizeHint = 128, log))
-                .flatten(FlattenStrategy.concat)
-              HttpEntity.Chunked(contentType, chunks)
-          }
-        })
-      }
+      Marshalling.WithOpenCharset(contentType.mediaType, { charset ⇒
+        value match {
+          case x: Multipart.Strict ⇒
+            val data = BodyPartRenderer.strict(x.strictParts, boundary, charset.nioCharset, partHeadersSizeHint = 128, log)
+            HttpEntity(contentType, data)
+          case _ ⇒
+            val chunks = value.parts
+              .transform("bodyPartRenderer", () ⇒ BodyPartRenderer.streamed(boundary, charset.nioCharset, partHeadersSizeHint = 128, log))
+              .flatten(FlattenStrategy.concat)
+            HttpEntity.Chunked(contentType, chunks)
+        }
+      })
     }
 }
 
diff --git a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToEntityMarshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToEntityMarshallers.scala
index 9e3145472d..0cb194783c 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToEntityMarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToEntityMarshallers.scala
@@ -59,9 +59,8 @@ trait PredefinedToEntityMarshallers extends MultipartMarshallers {
       HttpEntity(ContentType(`application/x-www-form-urlencoded`, charset), string)
     }
 
-  implicit val HttpEntityMarshaller: ToEntityMarshaller[MessageEntity] = Marshaller { value ⇒
-    // since we don't want to recode we simply ignore the charset determined by content negotiation here
-    FastFuture.successful(Marshalling.WithOpenCharset(value.contentType.mediaType, _ ⇒ value))
+  implicit val HttpEntityMarshaller: ToEntityMarshaller[MessageEntity] = Marshaller strict { value ⇒
+    Marshalling.WithFixedCharset(value.contentType.mediaType, value.contentType.charset, () ⇒ value)
   }
 }
 
diff --git a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToRequestMarshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToRequestMarshallers.scala
index 39a2801d25..352562ae09 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToRequestMarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToRequestMarshallers.scala
@@ -16,15 +16,15 @@ trait PredefinedToRequestMarshallers {
   implicit val fromRequest: TRM[HttpRequest] = Marshaller.opaque(identity)
 
   implicit def fromUri(implicit ec: ExecutionContext): TRM[Uri] =
-    Marshaller { uri ⇒ FastFuture.successful(Marshalling.Opaque(() ⇒ HttpRequest(uri = uri))) }
+    Marshaller strict { uri ⇒ Marshalling.Opaque(() ⇒ HttpRequest(uri = uri)) }
 
   implicit def fromMethodAndUriAndValue[S, T](implicit mt: ToEntityMarshaller[T],
                                               ec: ExecutionContext): TRM[(HttpMethod, Uri, T)] =
-    fromMethodAndUriAndHeadersAndValue[T].compose { case (m, u, v) ⇒ (m, u, Nil, v) }
+    fromMethodAndUriAndHeadersAndValue[T] compose { case (m, u, v) ⇒ (m, u, Nil, v) }
 
   implicit def fromMethodAndUriAndHeadersAndValue[T](implicit mt: ToEntityMarshaller[T],
                                                      ec: ExecutionContext): TRM[(HttpMethod, Uri, immutable.Seq[HttpHeader], T)] =
-    Marshaller { case (m, u, h, v) ⇒ mt(v).fast.map(_ map (HttpRequest(m, u, h, _))) }
+    Marshaller { case (m, u, h, v) ⇒ mt(v).fast map (_ map (_ map (HttpRequest(m, u, h, _)))) }
 }
 
 object PredefinedToRequestMarshallers extends PredefinedToRequestMarshallers
diff --git a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToResponseMarshallers.scala b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToResponseMarshallers.scala
index 19cf405aa8..70ca740377 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/PredefinedToResponseMarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/PredefinedToResponseMarshallers.scala
@@ -34,7 +34,7 @@ trait PredefinedToResponseMarshallers extends LowPriorityToResponseMarshallerImp
 
   implicit def fromStatusCodeAndHeadersAndValue[T](implicit mt: ToEntityMarshaller[T],
                                                    ec: ExecutionContext): TRM[(StatusCode, immutable.Seq[HttpHeader], T)] =
-    Marshaller { case (status, headers, value) ⇒ mt(value).fast.map(_ map (HttpResponse(status, headers, _))) }
+    Marshaller { case (status, headers, value) ⇒ mt(value).fast map (_ map (_ map (HttpResponse(status, headers, _)))) }
 }
 
 trait LowPriorityToResponseMarshallerImplicits {
diff --git a/akka-http/src/main/scala/akka/http/marshalling/ToResponseMarshallable.scala b/akka-http/src/main/scala/akka/http/marshalling/ToResponseMarshallable.scala
index c58e66dd59..b79d78d112 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/ToResponseMarshallable.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/ToResponseMarshallable.scala
@@ -18,15 +18,13 @@ trait ToResponseMarshallable {
 }
 
 object ToResponseMarshallable {
-  implicit def isMarshallable[A](_value: A)(implicit _marshaller: ToResponseMarshaller[A]): ToResponseMarshallable =
+  implicit def apply[A](_value: A)(implicit _marshaller: ToResponseMarshaller[A]): ToResponseMarshallable =
     new ToResponseMarshallable {
       type T = A
       def value: T = _value
       def marshaller: ToResponseMarshaller[T] = _marshaller
     }
 
-  implicit def marshallableIsMarshallable: ToResponseMarshaller[ToResponseMarshallable] =
-    Marshaller[ToResponseMarshallable, HttpResponse] { value ⇒
-      value.marshaller(value.value)
-    }
+  implicit val marshaller: ToResponseMarshaller[ToResponseMarshallable] =
+    Marshaller { marshallable ⇒ marshallable.marshaller(marshallable.value) }
 }
\ No newline at end of file
diff --git a/akka-http/src/main/scala/akka/http/marshalling/package.scala b/akka-http/src/main/scala/akka/http/marshalling/package.scala
index 2eec53b722..ceec008e3f 100644
--- a/akka-http/src/main/scala/akka/http/marshalling/package.scala
+++ b/akka-http/src/main/scala/akka/http/marshalling/package.scala
@@ -12,8 +12,4 @@ package object marshalling {
   type ToHeadersAndEntityMarshaller[T] = Marshaller[T, (immutable.Seq[HttpHeader], MessageEntity)]
   type ToResponseMarshaller[T] = Marshaller[T, HttpResponse]
   type ToRequestMarshaller[T] = Marshaller[T, HttpRequest]
-
-  type ToEntityMarshallers[T] = Marshallers[T, MessageEntity]
-  type ToResponseMarshallers[T] = Marshallers[T, HttpResponse]
-  type ToRequestMarshallers[T] = Marshallers[T, HttpRequest]
 }
diff --git a/akka-http/src/main/scala/akka/http/unmarshalling/GenericUnmarshallers.scala b/akka-http/src/main/scala/akka/http/unmarshalling/GenericUnmarshallers.scala
index ad12bea60a..58a8d42e34 100644
--- a/akka-http/src/main/scala/akka/http/unmarshalling/GenericUnmarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/unmarshalling/GenericUnmarshallers.scala
@@ -9,6 +9,8 @@ import akka.http.util.FastFuture
 
 trait GenericUnmarshallers extends LowerPriorityGenericUnmarshallers {
 
+  implicit def liftToTargetOptionUnmarshaller[A, B](um: Unmarshaller[A, B])(implicit ec: ExecutionContext): Unmarshaller[A, Option[B]] =
+    targetOptionUnmarshaller(um, ec)
   implicit def targetOptionUnmarshaller[A, B](implicit um: Unmarshaller[A, B], ec: ExecutionContext): Unmarshaller[A, Option[B]] =
     um map (Some(_)) withDefaultValue None
 }
@@ -18,12 +20,11 @@ sealed trait LowerPriorityGenericUnmarshallers {
   implicit def messageUnmarshallerFromEntityUnmarshaller[T](implicit um: FromEntityUnmarshaller[T]): FromMessageUnmarshaller[T] =
     Unmarshaller { request ⇒ um(request.entity) }
 
-  implicit def liftToSourceOptionUnmarshaller[A, B](um: Unmarshaller[A, B])(implicit ec: ExecutionContext): Unmarshaller[Option[A], B] =
-    sourceOptionUnmarshaller(um, ec)
-
-  implicit def sourceOptionUnmarshaller[A, B](implicit um: Unmarshaller[A, B], ec: ExecutionContext): Unmarshaller[Option[A], B] =
+  implicit def liftToSourceOptionUnmarshaller[A, B](um: Unmarshaller[A, B]): Unmarshaller[Option[A], B] =
+    sourceOptionUnmarshaller(um)
+  implicit def sourceOptionUnmarshaller[A, B](implicit um: Unmarshaller[A, B]): Unmarshaller[Option[A], B] =
     Unmarshaller {
       case Some(a) ⇒ um(a)
-      case None    ⇒ FastFuture.failed(UnmarshallingError.ContentExpected)
+      case None    ⇒ FastFuture.failed(Unmarshaller.NoContentException)
     }
 }
\ No newline at end of file
diff --git a/akka-http/src/main/scala/akka/http/unmarshalling/MultipartUnmarshallers.scala b/akka-http/src/main/scala/akka/http/unmarshalling/MultipartUnmarshallers.scala
index ba3b7ab6ac..e343dd71bf 100644
--- a/akka-http/src/main/scala/akka/http/unmarshalling/MultipartUnmarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/unmarshalling/MultipartUnmarshallers.scala
@@ -61,7 +61,7 @@ trait MultipartUnmarshallers {
       if (entity.contentType.mediaType.isMultipart && mediaRange.matches(entity.contentType.mediaType)) {
         entity.contentType.mediaType.params.get("boundary") match {
           case None ⇒
-            FastFuture.failed(UnmarshallingError.InvalidContent("Content-Type with a multipart media type must have a 'boundary' parameter"))
+            FastFuture.failed(new RuntimeException("Content-Type with a multipart media type must have a 'boundary' parameter"))
           case Some(boundary) ⇒
             import BodyPartParser._
             val parser = new BodyPartParser(defaultContentType, boundary, log)
@@ -96,7 +96,7 @@ trait MultipartUnmarshallers {
               }
             }
         }
-      } else FastFuture.failed(UnmarshallingError.UnsupportedContentType(ContentTypeRange(mediaRange) :: Nil))
+      } else FastFuture.failed(Unmarshaller.UnsupportedContentTypeException(mediaRange))
     }
 }
 
diff --git a/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromEntityUnmarshallers.scala b/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromEntityUnmarshallers.scala
index 9d1775e8d0..614ead66b1 100644
--- a/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromEntityUnmarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromEntityUnmarshallers.scala
@@ -38,9 +38,12 @@ trait PredefinedFromEntityUnmarshallers extends MultipartUnmarshallers {
       bytes.decodeString(entity.contentType.charset.nioCharset.name) // ouch!!!
     }
 
-  implicit def urlEncodedFormDataUnmarshaller(implicit fm: FlowMaterializer,
-                                              ec: ExecutionContext): FromEntityUnmarshaller[FormData] =
-    stringUnmarshaller mapWithInput { (entity, string) ⇒
+  implicit def defaultUrlEncodedFormDataUnmarshaller(implicit fm: FlowMaterializer,
+                                                     ec: ExecutionContext): FromEntityUnmarshaller[FormData] =
+    urlEncodedFormDataUnmarshaller(MediaTypes.`application/x-www-form-urlencoded`)
+  def urlEncodedFormDataUnmarshaller(ranges: ContentTypeRange*)(implicit fm: FlowMaterializer,
+                                                                ec: ExecutionContext): FromEntityUnmarshaller[FormData] =
+    stringUnmarshaller.forContentTypes(ranges: _*).mapWithInput { (entity, string) ⇒
       try {
         val nioCharset = entity.contentType.definedCharset.getOrElse(HttpCharsets.`UTF-8`).nioCharset
         val query = Uri.Query(string, nioCharset)
diff --git a/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromStringUnmarshallers.scala b/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromStringUnmarshallers.scala
index a6fd4c1756..e826885b3a 100644
--- a/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromStringUnmarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/unmarshalling/PredefinedFromStringUnmarshallers.scala
@@ -60,13 +60,14 @@ trait PredefinedFromStringUnmarshallers {
     string.toLowerCase match {
       case "true" | "yes" | "on"  ⇒ true
       case "false" | "no" | "off" ⇒ false
-      case x                      ⇒ throw UnmarshallingError.InvalidContent(s"'$x' is not a valid Boolean value")
+      case ""                     ⇒ throw Unmarshaller.NoContentException
+      case x                      ⇒ sys.error(s"'$x' is not a valid Boolean value")
     }
   }
 
   private def numberFormatError(value: String, target: String): PartialFunction[Throwable, Nothing] = {
     case e: NumberFormatException ⇒
-      throw UnmarshallingError.InvalidContent(s"'$value' is not a valid $target value", e)
+      throw if (value.isEmpty) Unmarshaller.NoContentException else new RuntimeException(s"'$value' is not a valid $target value", e)
   }
 }
 
diff --git a/akka-http/src/main/scala/akka/http/unmarshalling/Unmarshaller.scala b/akka-http/src/main/scala/akka/http/unmarshalling/Unmarshaller.scala
index 4011a90686..94852f72cb 100644
--- a/akka-http/src/main/scala/akka/http/unmarshalling/Unmarshaller.scala
+++ b/akka-http/src/main/scala/akka/http/unmarshalling/Unmarshaller.scala
@@ -4,11 +4,10 @@
 
 package akka.http.unmarshalling
 
-import scala.util.control.NonFatal
-import scala.collection.immutable
+import scala.util.control.{ NoStackTrace, NonFatal }
 import scala.concurrent.{ Future, ExecutionContext }
 import akka.http.util._
-import akka.http.model.{ HttpCharset, MediaType, ContentTypeRange }
+import akka.http.model._
 import FastFuture._
 
 trait Unmarshaller[-A, B] extends (A ⇒ Future[B]) {
@@ -26,7 +25,7 @@ trait Unmarshaller[-A, B] extends (A ⇒ Future[B]) {
     transform(_.fast recover pf)
 
   def withDefaultValue[BB >: B](defaultValue: BB)(implicit ec: ExecutionContext): Unmarshaller[A, BB] =
-    recover { case UnmarshallingError.ContentExpected ⇒ defaultValue }
+    recover { case Unmarshaller.NoContentException ⇒ defaultValue }
 }
 
 object Unmarshaller
@@ -34,15 +33,37 @@ object Unmarshaller
   with PredefinedFromEntityUnmarshallers
   with PredefinedFromStringUnmarshallers {
 
+  /**
+   * Creates an `Unmarshaller` from the given function.
+   */
   def apply[A, B](f: A ⇒ Future[B]): Unmarshaller[A, B] =
     new Unmarshaller[A, B] {
       def apply(a: A) =
         try f(a)
-        catch { case NonFatal(e) ⇒ FastFuture.failed(UnmarshallingError.InvalidContent(e.getMessage.nullAsEmpty, e)) }
+        catch { case NonFatal(e) ⇒ FastFuture.failed(e) }
     }
 
+  /**
+   * Helper for creating a synchronous `Unmarshaller` from the given function.
+   */
   def strict[A, B](f: A ⇒ B): Unmarshaller[A, B] = Unmarshaller(a ⇒ FastFuture.successful(f(a)))
 
+  /**
+   * Helper for creating a "super-unmarshaller" from a sequence of "sub-unmarshallers", which are tried
+   * in the given order. The first successful unmarshalling of a "sub-unmarshallers" is the one produced by the
+   * "super-unmarshaller".
+   */
+  def firstOf[A, B](unmarshallers: Unmarshaller[A, B]*)(implicit ec: ExecutionContext): Unmarshaller[A, B] =
+    Unmarshaller { a ⇒
+      def rec(ix: Int, supported: Set[ContentTypeRange]): Future[B] =
+        if (ix < unmarshallers.size) {
+          unmarshallers(ix)(a).fast.recoverWith {
+            case Unmarshaller.UnsupportedContentTypeException(supp) ⇒ rec(ix + 1, supported ++ supp)
+          }
+        } else FastFuture.failed(Unmarshaller.UnsupportedContentTypeException(supported))
+      rec(0, Set.empty)
+    }
+
   implicit def identityUnmarshaller[T]: Unmarshaller[T, T] = Unmarshaller(FastFuture.successful)
 
   // we don't define these methods directly on `Unmarshaller` due to variance constraints
@@ -54,31 +75,44 @@ object Unmarshaller
       Unmarshaller(a ⇒ um(a).fast.flatMap(f(a, _)))
   }
 
-  implicit class EnhancedToEntityUnmarshaller[T](val um: FromEntityUnmarshaller[T]) extends AnyVal {
-    def mapWithCharset[U](f: (T, HttpCharset) ⇒ U)(implicit ec: ExecutionContext): FromEntityUnmarshaller[U] =
-      um.mapWithInput { (entity, data) ⇒ f(data, entity.contentType.charset) }
+  implicit class EnhancedFromEntityUnmarshaller[A](val underlying: FromEntityUnmarshaller[A]) extends AnyVal {
+    def mapWithCharset[B](f: (A, HttpCharset) ⇒ B)(implicit ec: ExecutionContext): FromEntityUnmarshaller[B] =
+      underlying.mapWithInput { (entity, data) ⇒ f(data, entity.contentType.charset) }
 
-    def filterMediaType(allowed: MediaType*)(implicit ec: ExecutionContext): FromEntityUnmarshaller[T] =
-      um.flatMapWithInput { (entity, data) ⇒
-        if (allowed contains entity.contentType.mediaType) Future.successful(data)
-        else FastFuture.failed(UnmarshallingError.UnsupportedContentType(allowed map (ContentTypeRange(_)) toList))
+    /**
+     * Modifies the underlying [[Unmarshaller]] to only accept content-types matching one of the given ranges.
+     * If the underlying [[Unmarshaller]] already contains a content-type filter (also wrapped at some level),
+     * this filter is *replaced* by this method, not stacked!
+     */
+    def forContentTypes(ranges: ContentTypeRange*)(implicit ec: ExecutionContext): FromEntityUnmarshaller[A] =
+      Unmarshaller { entity ⇒
+        if (entity.contentType == ContentTypes.NoContentType || ranges.exists(_ matches entity.contentType)) {
+          underlying(entity).fast recoverWith retryWithPatchedContentType(underlying, entity)
+        } else FastFuture.failed(UnsupportedContentTypeException(ranges: _*))
       }
   }
-}
 
-sealed abstract class UnmarshallingError(msg: String, cause: Throwable = null) extends RuntimeException(msg, cause)
+  // must be moved out of the the [[EnhancedFromEntityUnmarshaller]] value class due to bug in scala 2.10:
+  // https://issues.scala-lang.org/browse/SI-8018
+  private def retryWithPatchedContentType[T](underlying: FromEntityUnmarshaller[T],
+                                             entity: HttpEntity): PartialFunction[Throwable, Future[T]] = {
+    case UnsupportedContentTypeException(supported) ⇒ underlying(entity withContentType supported.head.specimen)
+  }
 
-object UnmarshallingError {
-  case object ContentExpected
-    extends UnmarshallingError("Content expected")
+  /**
+   * Signals that unmarshalling failed because the entity was unexpectedly empty.
+   */
+  case object NoContentException extends RuntimeException("Message entity must not be empty") with NoStackTrace
 
-  final case class UnsupportedContentType(supported: immutable.Seq[ContentTypeRange])
-    extends UnmarshallingError(supported.mkString("Unsupported Content-Type, supported: ", ", ", ""))
+  /**
+   * Signals that unmarshalling failed because the entity content-type did not match one of the supported ranges.
+   * This error cannot be thrown by custom code, you need to use the `forContentTypes` modifier on a base
+   * [[akka.http.unmarshalling.Unmarshaller]] instead.
+   */
+  final case class UnsupportedContentTypeException(supported: Set[ContentTypeRange])
+    extends RuntimeException(supported.mkString("Unsupported Content-Type, supported: ", ", ", ""))
 
-  final case class InvalidContent(errorMessage: String, cause: Option[Throwable] = None)
-    extends UnmarshallingError(errorMessage, cause.orNull)
-
-  object InvalidContent {
-    def apply(errorMessage: String, cause: Throwable) = new InvalidContent(errorMessage, Some(cause))
+  object UnsupportedContentTypeException {
+    def apply(supported: ContentTypeRange*): UnsupportedContentTypeException = UnsupportedContentTypeException(Set(supported: _*))
   }
 }