+doc #20192 explain need of draining entities in server/client HTTP (#20852)

* +doc #20192 explain need of draining entities in server/client HTTP

* missing javadsl for Connection header

* Update HttpClientExampleDocTest.java

akka-docs/rst/scala/code/docs/http/scaladsl/HttpClientExampleSpec.scala

@@ -4,13 +4,110 @@
 
 package docs.http.scaladsl
 
+import akka.Done
 import akka.actor.{ ActorLogging, ActorSystem }
+import akka.http.scaladsl.model.HttpEntity.Strict
+import akka.http.scaladsl.model.HttpMessage.DiscardedEntity
+import akka.stream.{ IOResult, Materializer }
+import akka.stream.scaladsl.{ Framing, Sink }
 import akka.util.ByteString
 import docs.CompileOnlySpec
 import org.scalatest.{ Matchers, WordSpec }
 
+import scala.concurrent.{ ExecutionContextExecutor, Future }
+
 class HttpClientExampleSpec extends WordSpec with Matchers with CompileOnlySpec {
 
+  "manual-entity-consume-example-1" in compileOnlySpec {
+    //#manual-entity-consume-example-1
+    import java.io.File
+    import akka.actor.ActorSystem
+    import akka.stream.ActorMaterializer
+    import akka.stream.scaladsl.Framing
+    import akka.stream.scaladsl.FileIO
+    import akka.http.scaladsl.model._
+
+    implicit val system = ActorSystem()
+    implicit val dispatcher = system.dispatcher
+    implicit val materializer = ActorMaterializer()
+
+    val response: HttpResponse = ???
+
+    response.entity.dataBytes
+      .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 256))
+      .map(transformEachLine)
+      .runWith(FileIO.toPath(new File("/tmp/example.out").toPath))
+
+    def transformEachLine(line: ByteString): ByteString = ???
+
+    //#manual-entity-consume-example-1
+  }
+
+  "manual-entity-consume-example-2" in compileOnlySpec {
+    //#manual-entity-consume-example-2
+    import java.io.File
+    import akka.actor.ActorSystem
+    import akka.stream.ActorMaterializer
+    import akka.http.scaladsl.model._
+    import scala.concurrent.duration._
+
+    implicit val system = ActorSystem()
+    implicit val dispatcher = system.dispatcher
+    implicit val materializer = ActorMaterializer()
+
+    case class ExamplePerson(name: String)
+    def parse(line: ByteString): ExamplePerson = ???
+
+    val response: HttpResponse = ???
+
+    // toStrict to enforce all data be loaded into memory from the connection
+    val strictEntity: Future[HttpEntity.Strict] = response.entity.toStrict(3.seconds)
+
+    // while API remains the same to consume dataBytes, now they're in memory already:
+    val transformedData: Future[ExamplePerson] =
+      strictEntity flatMap { e =>
+        e.dataBytes
+          .runFold(ByteString.empty) { case (acc, b) => acc ++ b }
+          .map(parse)
+      }
+
+    //#manual-entity-consume-example-2
+  }
+
+  "manual-entity-discard-example-1" in compileOnlySpec {
+    //#manual-entity-discard-example-1
+    import akka.actor.ActorSystem
+    import akka.stream.ActorMaterializer
+    import akka.http.scaladsl.model._
+
+    implicit val system = ActorSystem()
+    implicit val dispatcher = system.dispatcher
+    implicit val materializer = ActorMaterializer()
+
+    val response1: HttpResponse = ??? // obtained from an HTTP call (see examples below)
+
+    val discarded: DiscardedEntity = response1.discardEntityBytes()
+    discarded.future.onComplete { case done => println("Entity discarded completely!") }
+
+    //#manual-entity-discard-example-1
+  }
+  "manual-entity-discard-example-2" in compileOnlySpec {
+    import akka.actor.ActorSystem
+    import akka.stream.ActorMaterializer
+    import akka.http.scaladsl.model._
+
+    implicit val system = ActorSystem()
+    implicit val dispatcher = system.dispatcher
+    implicit val materializer = ActorMaterializer()
+
+    //#manual-entity-discard-example-2
+    val response1: HttpResponse = ??? // obtained from an HTTP call (see examples below)
+
+    val discardingComplete: Future[Done] = response1.entity.dataBytes.runWith(Sink.ignore)
+    discardingComplete.onComplete { case done => println("Entity discarded completely!") }
+    //#manual-entity-discard-example-2
+  }
+
   "outgoing-connection-example" in compileOnlySpec {
     //#outgoing-connection-example
     import akka.actor.ActorSystem
@@ -74,6 +171,7 @@ class HttpClientExampleSpec extends WordSpec with Matchers with CompileOnlySpec
     import akka.stream.ActorMaterializer
 
     import scala.concurrent.Future
+    import scala.util.{ Failure, Success }
 
     implicit val system = ActorSystem()
     implicit val materializer = ActorMaterializer()

akka-docs/rst/scala/code/docs/http/scaladsl/HttpServerExampleSpec.scala

@@ -5,11 +5,13 @@
 package docs.http.scaladsl
 
 import akka.event.LoggingAdapter
+import akka.http.scaladsl.model.{ RequestEntity, StatusCodes }
+import akka.stream.scaladsl.Sink
 import akka.testkit.TestActors
 import docs.CompileOnlySpec
 import org.scalatest.{ Matchers, WordSpec }
-import scala.language.postfixOps
 
+import scala.language.postfixOps
 import scala.concurrent.{ ExecutionContext, Future }
 
 class HttpServerExampleSpec extends WordSpec with Matchers
@@ -159,7 +161,7 @@ class HttpServerExampleSpec extends WordSpec with Matchers
     val httpEcho = Flow[HttpRequest]
       .via(reactToConnectionFailure)
       .map { request =>
-        // simple text "echo" response:
+        // simple streaming (!) "echo" response:
         HttpResponse(entity = HttpEntity(ContentTypes.`text/plain(UTF-8)`, request.entity.dataBytes))
       }
 
@@ -195,7 +197,8 @@ class HttpServerExampleSpec extends WordSpec with Matchers
       case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
         sys.error("BOOM!")
 
-      case _: HttpRequest =>
+      case r: HttpRequest =>
+        r.discardEntityBytes() // important to drain incoming HTTP Entity stream
         HttpResponse(404, entity = "Unknown resource!")
     }
 
@@ -237,7 +240,8 @@ class HttpServerExampleSpec extends WordSpec with Matchers
           case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
             sys.error("BOOM!")
 
-          case _: HttpRequest =>
+          case r: HttpRequest =>
+            r.discardEntityBytes() // important to drain incoming HTTP Entity stream
             HttpResponse(404, entity = "Unknown resource!")
         }
 
@@ -553,6 +557,126 @@ class HttpServerExampleSpec extends WordSpec with Matchers
     }
     //#actor-interaction
   }
+  
+  "consume entity using entity directive" in compileOnlySpec {
+    //#consume-entity-directive
+    import akka.actor.ActorSystem
+    import akka.http.scaladsl.server.Directives._
+    import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+    import akka.stream.ActorMaterializer
+    import spray.json.DefaultJsonProtocol._
+
+    implicit val system = ActorSystem()
+    implicit val materializer = ActorMaterializer()
+    // needed for the future flatMap/onComplete in the end
+    implicit val executionContext = system.dispatcher
+
+    final case class Bid(userId: String, bid: Int)
+
+    // these are from spray-json
+    implicit val bidFormat = jsonFormat2(Bid)
+    
+    val route =
+      path("bid") {
+        put {
+          entity(as[Bid]) { bid =>
+            // incoming entity is fully consumed and converted into a Bid
+            complete("The bid was: " + bid)
+          }
+        }
+      }
+    //#consume-entity-directive
+  }
+  
+  "consume entity using raw dataBytes to file" in compileOnlySpec {
+    //#consume-raw-dataBytes
+    import akka.actor.ActorSystem
+    import akka.stream.scaladsl.FileIO
+    import akka.http.scaladsl.server.Directives._
+    import akka.stream.ActorMaterializer
+    import java.io.File
+
+    implicit val system = ActorSystem()
+    implicit val materializer = ActorMaterializer()
+    // needed for the future flatMap/onComplete in the end
+    implicit val executionContext = system.dispatcher
+
+    val route =
+      (put & path("lines")) {
+        withoutSizeLimit {
+          extractDataBytes { bytes =>
+            val finishedWriting = bytes.runWith(FileIO.toPath(new File("/tmp/example.out").toPath))
+            
+            // we only want to respond once the incoming data has been handled:
+            onComplete(finishedWriting) { ioResult =>
+              complete("Finished writing data: " + ioResult)
+            }
+          }
+        }
+      }
+    //#consume-raw-dataBytes
+  }
+  
+  "drain entity using request#discardEntityBytes" in compileOnlySpec {
+    //#discard-discardEntityBytes
+    import akka.actor.ActorSystem
+    import akka.stream.scaladsl.FileIO
+    import akka.http.scaladsl.server.Directives._
+    import akka.stream.ActorMaterializer
+    import akka.http.scaladsl.model.HttpRequest
+
+    implicit val system = ActorSystem()
+    implicit val materializer = ActorMaterializer()
+    // needed for the future flatMap/onComplete in the end
+    implicit val executionContext = system.dispatcher
+
+    val route =
+      (put & path("lines")) {
+        withoutSizeLimit {
+          extractRequest { r: HttpRequest =>
+            val finishedWriting = r.discardEntityBytes().future
+            
+            // we only want to respond once the incoming data has been handled:
+            onComplete(finishedWriting) { done =>
+              complete("Drained all data from connection... (" + done + ")")
+            }
+          }
+        }
+      }
+    //#discard-discardEntityBytes
+  }
+  
+  "discard entity manually" in compileOnlySpec {
+    //#discard-close-connections
+    import akka.actor.ActorSystem
+    import akka.stream.scaladsl.Sink
+    import akka.http.scaladsl.server.Directives._
+    import akka.http.scaladsl.model.headers.Connection
+    import akka.stream.ActorMaterializer
+
+    implicit val system = ActorSystem()
+    implicit val materializer = ActorMaterializer()
+    // needed for the future flatMap/onComplete in the end
+    implicit val executionContext = system.dispatcher
+
+    val route =
+      (put & path("lines")) {
+        withoutSizeLimit {
+          extractDataBytes { data => 
+            // Closing connections, method 1 (eager):
+            // we deem this request as illegal, and close the connection right away:
+            data.runWith(Sink.cancelled) // "brutally" closes the connection
+            
+            // Closing connections, method 2 (graceful):
+            // consider draining connection and replying with `Connection: Close` header
+            // if you want the client to close after this request/reply cycle instead:
+            respondWithHeader(Connection("close"))
+            complete(StatusCodes.Forbidden -> "Not allowed!")
+          }
+        }
+      }
+    //#discard-close-connections
+  }
 
 
 }

akka-docs/rst/scala/code/docs/http/scaladsl/server/WebSocketExampleSpec.scala

@@ -49,7 +49,9 @@ class WebSocketExampleSpec extends WordSpec with Matchers with CompileOnlySpec {
           case Some(upgrade) => upgrade.handleMessages(greeterWebSocketService)
           case None          => HttpResponse(400, entity = "Not a valid websocket request!")
         }
-      case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
+      case r: HttpRequest =>
+        r.discardEntityBytes() // important to drain incoming HTTP Entity stream
+        HttpResponse(404, entity = "Unknown resource!")
     }
     //#websocket-request-handling
 
@@ -84,6 +86,7 @@ class WebSocketExampleSpec extends WordSpec with Matchers with CompileOnlySpec {
         .collect {
           case tm: TextMessage => TextMessage(Source.single("Hello ") ++ tm.textStream)
           // ignore binary messages
+          // TODO #20096 in case a Streamed message comes in, we should runWith(Sink.ignore) its data
         }
 
     //#websocket-routing

akka-docs/rst/scala/code/docs/http/scaladsl/server/directives/RouteDirectivesExamplesSpec.scala

@@ -22,10 +22,10 @@ class RouteDirectivesExamplesSpec extends RoutingSpec {
           complete(StatusCodes.OK)
         } ~
         path("c") {
-          complete(StatusCodes.Created, "bar")
+          complete(StatusCodes.Created -> "bar")
         } ~
         path("d") {
-          complete(201, "bar")
+          complete(201 -> "bar")
         } ~
         path("e") {
           complete(StatusCodes.Created, List(`Content-Type`(`text/plain(UTF-8)`)), "bar")