=core akka#424 add full example with custom content type

akka-http-marshallers-scala/akka-http-spray-json/src/main/mima-filters/10.0.1.backwards.excludes

@@ -3,3 +3,7 @@
 # Code compiled with 10.0.0 would have to be a class extending from SprayJsonSupport that is then used in 10.0.1 code
 # and would actually use the new implicit. Unlikely pattern.
 ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.sprayJsValueByteStringUnmarshaller")
+
+# =core #424 fix stream marshalling, better errors, more examples 
+# added new implicit marshaller provider; old method became non-implicit so binary compatibility is kept, but new code will use the new implicit
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.scaladsl.marshalling.LowPriorityToResponseMarshallerImplicits.fromEntityStreamingSupportAndEntityMarshaller")

akka-http-marshallers-scala/akka-http-spray-json/src/main/scala/akka/http/scaladsl/marshallers/sprayjson/SprayJsonSupport.scala

@@ -70,7 +70,9 @@ trait SprayJsonSupport {
 object SprayJsonSupport extends SprayJsonSupport
 
 /**
- * Entity streaming support, independent of used Json parsing library etc.
+ * Entity streaming support, implemented using spray-json.
+ *
+ * See also <a href="https://github.com/spray/spray-json">github.com/spray/spray-json</a> for details about Spray JSON itself
  */
 object SprayJsonEntityStreamingSupport {
 
@@ -112,8 +114,6 @@ final class JsonEntityStreamingSupport private[akka] (
 ) extends common.JsonEntityStreamingSupport {
   import akka.http.impl.util.JavaMapping.Implicits._
 
-  println("this = " + this)
-
   def this(maxObjectSize: Int) =
     this(
       maxObjectSize,

akka-http/src/main/scala/akka/http/scaladsl/marshalling/PredefinedToResponseMarshallers.scala

@@ -100,12 +100,12 @@ trait LowPriorityToResponseMarshallerImplicits {
   implicit def liftMarshaller[T](implicit m: ToEntityMarshaller[T]): ToResponseMarshaller[T] =
     PredefinedToResponseMarshallers.fromToEntityMarshaller()
 
+  @deprecated("This method exists only for the purpose of binary compatibility, it used to be implicit.", "10.0.2")
+  def fromEntityStreamingSupportAndEntityMarshaller[T, M](s: EntityStreamingSupport, m: ToEntityMarshaller[T]): ToResponseMarshaller[Source[T, M]] =
+    fromEntityStreamingSupportAndEntityMarshaller(s, m, null)
+
   // FIXME deduplicate this!!!
-  implicit def fromEntityStreamingSupportAndEntityMarshaller[T, M](
-    implicit
-    s:   EntityStreamingSupport,
-    m:   ToEntityMarshaller[T],
-    tag: ClassTag[T]            = null): ToResponseMarshaller[Source[T, M]] = {
+  implicit def fromEntityStreamingSupportAndEntityMarshaller[T, M](implicit s: EntityStreamingSupport, m: ToEntityMarshaller[T], tag: ClassTag[T]): ToResponseMarshaller[Source[T, M]] = {
     Marshaller[Source[T, M], HttpResponse] { implicit ec ⇒ source ⇒
       FastFuture successful {
         Marshalling.WithFixedContentType(s.contentType, () ⇒ {

docs/src/main/paradox/scala/http/routing-dsl/source-streaming-support.md

@@ -115,3 +115,14 @@ instances, which can be provided using any JSON marshalling library (such as Cir
 
 When implementing a custom support trait, one should simply extend the `EntityStreamingSupport` abstract class,
 and implement all of it's methods. It's best to use the existing implementations as a guideline.
+
+## Supporting custom content types
+
+In order to marshal into custom content types both a `Marshaller` that can handle that content type, 
+**as well as an `EntityStreamingSupport` of matching content type** is required.
+
+Refer to the below, complete, example showcasing how to configure a custom marshaller as well as change 
+the entity streaming's support content type to be compatible. This is an area that would benefit from additional type-safety,
+which we hope to add in a future release.
+
+@@snip [JsonStreamingFullExamples.scala](../../../../../test/java/docs/http/scaladsl/server/JsonStreamingFullExamples.scala) { #custom-content-type }

docs/src/test/scala/docs/http/scaladsl/server/directives/JsonStreamingFullExamples.scala

@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 2009-2015 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.http.scaladsl.server.directives
+
+import akka.NotUsed
+import akka.http.scaladsl.common.{ EntityStreamingSupport, JsonEntityStreamingSupport }
+import akka.http.scaladsl.marshalling._
+import akka.http.scaladsl.model._
+import akka.http.scaladsl.model.headers.Accept
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.server.{ UnacceptedResponseContentTypeRejection, UnsupportedRequestContentTypeRejection }
+import akka.stream.scaladsl.{ Flow, Source }
+import akka.util.ByteString
+import docs.http.scaladsl.server.RoutingSpec
+import org.scalatest.{ FlatSpec, WordSpec }
+
+import scala.concurrent.Future
+
+class JsonStreamingFullExamples extends WordSpec {
+
+  "compile only spec" in {}
+
+  //#custom-content-type
+  import akka.NotUsed
+  import akka.actor.ActorSystem
+  import akka.http.scaladsl.Http
+  import akka.http.scaladsl.common.{ EntityStreamingSupport, JsonEntityStreamingSupport }
+  import akka.http.scaladsl.model.{ HttpEntity, StatusCodes, _ }
+  import akka.http.scaladsl.server.Directives._
+  import akka.stream.ActorMaterializer
+  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+  import akka.http.scaladsl.marshalling.{ Marshaller, ToEntityMarshaller, ToResponseMarshaller }
+  import akka.http.scaladsl.model.TransferEncodings.gzip
+  import akka.http.scaladsl.model.headers.{ HttpEncoding, HttpEncodings }
+  import akka.http.scaladsl.model.ws.{ Message, TextMessage }
+  import akka.stream.scaladsl.{ Flow, Source }
+  import akka.util.ByteString
+  import spray.json.DefaultJsonProtocol
+  import spray.json.DefaultJsonProtocol._
+
+  import scala.concurrent.Future
+  import scala.io.StdIn
+  import scala.util.Random
+
+  final case class User(name: String, id: String)
+
+  trait UserProtocol extends DefaultJsonProtocol {
+
+    import spray.json._
+
+    implicit val userFormat = jsonFormat2(User)
+
+    val `vnd.example.api.v1+json` =
+      MediaType.applicationWithFixedCharset("vnd.example.api.v1+json", HttpCharsets.`UTF-8`)
+    val ct = ContentType.apply(`vnd.example.api.v1+json`)
+
+    implicit def userMarshaller: ToEntityMarshaller[User] = Marshaller.oneOf(
+      Marshaller.withFixedContentType(`vnd.example.api.v1+json`) { organisation ⇒
+        HttpEntity(`vnd.example.api.v1+json`, organisation.toJson.compactPrint)
+      })
+  }
+
+  object ApiServer extends App with UserProtocol {
+    implicit val system = ActorSystem("api")
+    implicit val materializer = ActorMaterializer()
+    implicit val executionContext = system.dispatcher
+
+    implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()
+      .withContentType(ct)
+      .withParallelMarshalling(parallelism = 10, unordered = false)
+
+    // (fake) async database query api
+    def dummyUser(id: String) = User(s"User $id", id.toString)
+
+    def fetchUsers(): Source[User, NotUsed] = Source.fromIterator(() ⇒ Iterator.fill(10000) {
+      val id = Random.nextInt()
+      dummyUser(id.toString)
+    })
+
+    val route =
+      pathPrefix("users") {
+        get {
+          complete(fetchUsers())
+        }
+      }
+
+    val bindingFuture = Http().bindAndHandle(route, "localhost", 8080)
+
+    println(s"Server online at http://localhost:8080/\nPress RETURN to stop...")
+    StdIn.readLine()
+    bindingFuture.flatMap(_.unbind()).onComplete(_ ⇒ system.terminate())
+  }
+
+  //#custom-content-type
+}