Avoid eagerly failing connections when request closes (#2069)

* Avoid eagerly failing connections when request closes

* Test scalariform

akka-http-core/src/main/scala/akka/http/impl/engine/server/ServerTerminator.scala

@@ -228,6 +228,11 @@ private[http] final class GracefulTerminatorStage(settings: ServerSettings)
  pendingUserHandlerResponse = false
  push(toNet, response)
  }
+
+ override def onUpstreamFinish(): Unit = {
+ // don't finish the whole bidi stage, just propagate the completion:
+ complete(toNet)
+ }
  })
  setHandler(toUser, new OutHandler {
  override def onPull(): Unit = {
@@ -241,6 +246,11 @@ private[http] final class GracefulTerminatorStage(settings: ServerSettings)
  pendingUserHandlerResponse = true
  push(toUser, request)
  }
+
+ override def onUpstreamFinish(): Unit = {
+ // don't finish the whole bidi stage, just propagate the completion:
+ complete(toUser)
+ }
  })
  setHandler(toNet, new OutHandler {
  override def onPull(): Unit = {

akka-http-core/src/test/scala/akka/http/scaladsl/ClientServerSpec.scala

@@ -514,6 +514,7 @@ class ClientServerSpec extends WordSpec with Matchers with BeforeAndAfterAll wit
  connSourceSub.cancel()
  }
  }
+
  "complete a request/response when request has `Connection: close` set" in Utils.assertAllStagesStopped {
  // In akka/akka#19542 / akka/akka-http#459 it was observed that when an akka-http closes the connection after
  // a request, the TCP connection is sometimes aborted. Aborting means that `socket.close` is called with SO_LINGER = 0
@@ -557,6 +558,33 @@ class ClientServerSpec extends WordSpec with Matchers with BeforeAndAfterAll wit
  } finally server.foreach(_.unbind())
  }
 
+ "complete a request/response when the request side immediately closes the connection after sending the request" in Utils.assertAllStagesStopped {
+ val (hostname, port) = ("localhost", 8080)
+ val responsePromise = Promise[HttpResponse]()
+
+ // settings adapting network buffer sizes
+ val serverSettings = ServerSettings(system)
+
+ val server = Http().bindAndHandleAsync(_ ⇒ responsePromise.future, hostname, port, settings = serverSettings)
+
+ try {
+ val result = Source.single(ByteString(
+ """GET / HTTP/1.1
+Host: example.com
+
+"""))
+ .via(Tcp().outgoingConnection(hostname, port))
+ .runWith(Sink.reduce[ByteString](_ ++ _))
+ Try(Await.result(result, 2.seconds).utf8String) match {
+ case scala.util.Success(body) ⇒ fail(body)
+ case scala.util.Failure(_: TimeoutException) ⇒ // Expected
+ }
+ } finally {
+ responsePromise.failure(new TimeoutException())
+ server.foreach(_.unbind())
+ }
+ }
+
  "complete a request/response over https when request has `Connection: close` set" in Utils.assertAllStagesStopped {
  // akka/akka-http#1219
  val serverToClientNetworkBufferSize = 1000