Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

Test: Inspect chunked responses during termination #3139

Merged
merged 6 commits into from
Jun 3, 2020
Merged

Test: Inspect chunked responses during termination #3139

Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4726e5a
Inspect chunked responses during termination
ennru May 8, 2020
dd17282
A separate actor system for the client request
ennru May 11, 2020
ccf01b1
Use eventually
ennru May 15, 2020
6c41cad
Add ignored test for CloseDelimited
ennru May 15, 2020
b2d1fbb
Don't expect mulitple replies before termination
ennru Jun 3, 2020
ced41d4
Don't expect mulitple replies before termination
ennru Jun 3, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions akka-http-core/src/test/scala/akka/http/scaladsl/GracefulTerminationSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ package akka.http.scaladsl
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.{ ArrayBlockingQueue, TimeUnit }

import akka.Done
import akka.actor.ActorSystem
import akka.http.impl.util._
import akka.http.scaladsl.model.HttpEntity._
Expand All @@ -15,6 +16,8 @@ import akka.http.scaladsl.model.headers.Connection
import akka.http.scaladsl.settings.ClientConnectionSettings
import akka.http.scaladsl.settings.{ ConnectionPoolSettings, ServerSettings }
import akka.stream.scaladsl._
import akka.stream.testkit.TestSubscriber.{ OnError, OnNext }
import akka.stream.testkit.scaladsl.TestSink
import akka.stream.{ Server => _, _ }
import akka.testkit._
import akka.util.ByteString
Expand All @@ -41,6 +44,20 @@ class GracefulTerminationSpec

implicit override val patience = PatienceConfig(5.seconds.dilated(system), 200.millis)

"Unbinding" should {
"not allow new connections" in new TestSetup {
Await.result(serverBinding.unbind(), 1.second) should ===(Done)

// immediately trying a new connection should cause `Connection refused` since we unbind immediately:
val r = makeRequest(ensureNewConnection = true)
val ex = intercept[StreamTcpException] {
Await.result(r, 2.seconds)
}
ex.getMessage should include("Connection refused")
serverBinding.terminate(hardDeadline = 2.seconds)
}
}

"Graceful termination" should {

"stop accepting new connections" in new TestSetup {
Expand All @@ -60,6 +77,53 @@ class GracefulTerminationSpec
ex.getMessage should include("Connection refused")
}

"fail chunked response streams" in new TestSetup {
val clientSystem = ActorSystem("client")
val r1 =
Http()(clientSystem).singleRequest(nextRequest, connectionContext = clientConnectionContext, settings = basePoolSettings)

// reply with an infinite entity stream
val chunks = Source
.fromIterator(() => Iterator.from(1).map(v => ChunkStreamPart(s"reply$v,")))
.throttle(1, 300.millis)
reply(_ => HttpResponse(entity = HttpEntity.Chunked(ContentTypes.`text/plain(UTF-8)`, chunks)))

// start reading the response
val response = r1.futureValue.entity.dataBytes
.via(Framing.delimiter(ByteString(","), 20))
.runWith(TestSink.probe[ByteString])
response.requestNext().utf8String should ===("reply1")

try {
val termination = serverBinding.terminate(hardDeadline = 50.millis)
response.request(20)
// local testing shows the stream fails long after the 50 ms deadline
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So this test shows the deadline isn't actually enforced for streams like this?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(I guess that might not be a big problem since the actorsystem shutdown will shut down the materializer and fail the stream anyway?)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm banging my head against this a bit... What I'm seeing is that the GracefulTerminationStage does something after those 50ms but that does not kill the connection immediately (especially not the response stream). My hypothesis is, since the termination stage does not monitor response entity streams, it does something which will only indirectly stop the connection at some point. In any case, it does not cancel connection but complete it, which will still keep proper semantics in this case (because chunked encoding can detect truncation) but not in the below one.

response.expectNext().utf8String should ===("reply2")
response.expectNext().utf8String should ===("reply3")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we remove a couple of these as well?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

response.expectNext().utf8String should ===("reply4")
response.expectNext().utf8String should ===("reply5")
val e1 = response.expectEvent()
if (e1.isInstanceOf[OnNext[_]]) {
val e2 = response.expectEvent()
if (e2.isInstanceOf[OnNext[_]]) {
val e3 = response.expectEvent()
if (e3.isInstanceOf[OnNext[_]]) {
fail("the chunked entity stream is expected to fail")
} else if (!e3.isInstanceOf[OnError]) {
fail(s"the chunked entity stream is expected to fail, got $e3")
}
} else if (!e2.isInstanceOf[OnError]) {
fail(s"the chunked entity stream is expected to fail, got $e2")
}
} else if (!e1.isInstanceOf[OnError]) {
fail(s"the chunked entity stream is expected to fail, got $e1")
}
termination.futureValue shouldBe Http.HttpServerTerminated
} finally {
TestKit.shutdownActorSystem(clientSystem)
}
}

"provide whenTerminated future that completes once server has completed termination (no connections)" in new TestSetup {
val time: FiniteDuration = 2.seconds
val deadline: Deadline = time.fromNow
Expand Down