diff --git a/docs/src/main/paradox/client-side/configuration.md b/docs/src/main/paradox/client-side/configuration.md index 3e1066f89..609f6a944 100644 --- a/docs/src/main/paradox/client-side/configuration.md +++ b/docs/src/main/paradox/client-side/configuration.md @@ -20,6 +20,47 @@ Basic client settings can be overridden in multiple ways: @@snip [reference.conf](/http-core/src/main/resources/reference.conf) { #client-settings } +## Strict Response Entities + +By default response entities are streamed, so the application has to consume (or discard) each response entity before +the connection can be used for the next request. Setting `pekko.http.client.strict-response-entity-timeout` to a +duration makes the client collect every response entity into a strict entity (`HttpEntity.Strict`) before the response is +dispatched to the application: + +``` +pekko.http.client.strict-response-entity-timeout = 10s +``` + +Keep in mind that this buffers each complete response body in memory. A response that is not fully received within the +configured duration fails with a `TimeoutException`, and one that exceeds +`pekko.http.client.strict-response-entity-max-bytes` (8 MB by default) fails with an `EntityStreamException`; in both +cases the connection is failed. Trailing headers of chunked responses are dropped, as they are with +`HttpEntity.toStrict`. + +This applies to the HTTP/1.1 client, which includes the connection pool backing the +@ref[request-level](request-level.md) and @ref[host-level](host-level.md) APIs, and to the +@ref[HTTP/2 client](http2.md). + +### On HTTP/2 + +Responses on an HTTP/1.1 connection are sequential, so collecting one response entity never delays another. On HTTP/2 +several requests are in flight on one connection at the same time, which brings a few things to be aware of: + + * Entities are collected for up to `pekko.http.client.http2.max-concurrent-streams` responses concurrently, so that a + large response does not hold up smaller ones on other streams. + * Responses are emitted in the order in which their entities complete, not in the order in which their headers + arrived. HTTP/2 responses are unordered anyway and have to be correlated to their request via a + @apidoc[RequestResponseAssociation], so this does not break the API contract, but it does change observed ordering. + * Worst-case memory usage per connection is `max-concurrent-streams` × `strict-response-entity-max-bytes`, which is + 256 × 8 MB with the defaults. Tune both settings for the responses you actually expect. + * A response entity that times out or exceeds the maximum fails the whole connection, and with it every other stream + in flight on that connection. The remaining body cannot be skipped safely, so there is no way to fail only the one + response. With `Http().connectionTo(host).managedPersistentHttp2()` the connection is re-established afterwards + according to `pekko.http.client.http2.max-persistent-attempts`. + * Entity data is read from the network as fast as the peer sends it (up to the configured maximum) instead of at the + pace the application consumes it, so HTTP/2 flow control no longer reflects application backpressure for response + bodies. + ## Pool Settings Pool settings influence the behavior of client connection pools as used with APIs like `Http.singleRequest` diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes new file mode 100644 index 000000000..60ab6e579 --- /dev/null +++ b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes @@ -0,0 +1,21 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# optional strict response entities for the HTTP client https://github.com/apache/pekko-http/pull/1233 +ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.ClientConnectionSettings.withStrictResponseEntityMaxBytes") +ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ClientConnectionSettings.strictResponseEntityTimeout") +ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ClientConnectionSettings.strictResponseEntityMaxBytes") diff --git a/http-core/src/main/resources/reference.conf b/http-core/src/main/resources/reference.conf index 1c7019879..d87810e2c 100644 --- a/http-core/src/main/resources/reference.conf +++ b/http-core/src/main/resources/reference.conf @@ -447,6 +447,40 @@ pekko.http { max-content-length = infinite } + # If set to a duration, response entities are collected into strict entities + # (`HttpEntity.Strict`) before the response is dispatched to the application. + # `off` (the default) leaves response entities streamed. + # + # Note that enabling this means that the complete response body is buffered in + # memory. Responses that are not fully received within the given duration are + # failed with a `TimeoutException`, responses bigger than + # `strict-response-entity-max-bytes` with an `EntityStreamException`. In both + # cases the connection is failed as well, since the remaining response body + # cannot be skipped safely. + # + # Trailing headers of chunked responses are dropped, as they are with + # `HttpEntity.toStrict`. + # + # This applies to the HTTP/1.1 client (including the connection pool used by + # `singleRequest` and the host-level API) and to the HTTP/2 client. On HTTP/2 + # the entities of up to `pekko.http.client.http2.max-concurrent-streams` + # responses are collected concurrently, so that a big response does not delay + # smaller ones on other streams. Note that this means responses are emitted in + # the order in which their entities complete rather than in the order their + # headers arrived; HTTP/2 responses are unordered anyway and have to be + # correlated to their request via a `RequestResponseAssociation`. Also note + # that the worst case memory usage per HTTP/2 connection is + # `max-concurrent-streams` * `strict-response-entity-max-bytes`, and that + # failing one response entity fails the whole connection, and with it every + # other stream in flight on it. + strict-response-entity-timeout = off + + # The maximum number of bytes collected per response entity when + # `strict-response-entity-timeout` is enabled. Ignored otherwise. + # + # Set to `infinite` to only rely on `pekko.http.client.parsing.max-content-length`. + strict-response-entity-max-bytes = ${pekko.http.parsing.max-to-strict-bytes} + # Enables/disables the logging of unencrypted HTTP traffic to and from the HTTP # client for debugging reasons. # diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala index 6672d7446..df8356787 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala @@ -103,6 +103,12 @@ private[http] object OutgoingConnectionBlueprint { val responsePrep = Flow[List[ParserOutput.ResponseOutput]] .mapConcat(ConstantFun.scalaIdentityFunction) .via(new PrepareResponse(parserSettings)) + .via(settings.strictResponseEntityTimeout match { + case Some(timeout) => + // responses on an HTTP/1.1 connection are sequential, so collecting one entity never delays another + StreamUtils.strictifyResponseEntities(timeout, settings.strictResponseEntityMaxBytes, parallelism = 1) + case None => Flow[HttpResponse] + }) val terminationFanout = b.add(Broadcast[HttpResponse](2)) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala index 31454417f..9d0cb77b2 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala @@ -167,7 +167,27 @@ private[http] object Http2Blueprint { StreamUtils.statefulAttrsMap[Http2SubStream, HttpResponse] { attrs => val headerParser = masterHttpHeaderParser.createShallowCopy() stream => ResponseParsing.parseResponse(headerParser, settings.parserSettings, attrs)(stream) - }) + }.via(strictifyResponseEntitiesIfConfigured(settings))) + + /** + * Collects response entities into strict entities if `pekko.http.client.strict-response-entity-timeout` is + * configured. + * + * Entities are collected for up to `max-concurrent-streams` responses at a time so that a big response does not + * delay smaller ones on other streams. Because entities complete in an order of their own, this changes the order + * in which responses are emitted; HTTP/2 responses are unordered anyway and have to be correlated to their request + * via a `RequestResponseAssociation`. + */ + private def strictifyResponseEntitiesIfConfigured( + settings: ClientConnectionSettings): Flow[HttpResponse, HttpResponse, NotUsed] = + settings.strictResponseEntityTimeout match { + case Some(timeout) => + StreamUtils.strictifyResponseEntities( + timeout, + settings.strictResponseEntityMaxBytes, + parallelism = settings.http2Settings.maxConcurrentStreams) + case None => Flow[HttpResponse] + } def idleTimeoutIfConfigured(timeout: Duration): BidiFlow[ByteString, ByteString, ByteString, ByteString, NotUsed] = timeout match { diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala index 9da508eac..439c81602 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala @@ -44,6 +44,8 @@ private[pekko] final case class ClientConnectionSettingsImpl( socketOptions: immutable.Seq[SocketOption], parserSettings: ParserSettings, streamCancellationDelay: FiniteDuration, + strictResponseEntityTimeout: Option[FiniteDuration], + strictResponseEntityMaxBytes: Long, localAddress: Option[InetSocketAddress], http2Settings: Http2ClientSettings, transport: ClientTransport) @@ -51,6 +53,10 @@ private[pekko] final case class ClientConnectionSettingsImpl( require(connectingTimeout >= Duration.Zero, "connectingTimeout must be >= 0") require(requestHeaderSizeHint > 0, "request-size-hint must be > 0") + require( + strictResponseEntityTimeout.forall(_ > Duration.Zero), + "strict-response-entity-timeout must be > 0 or `off`") + require(strictResponseEntityMaxBytes >= 0, "strict-response-entity-max-bytes must be >= 0") require( Try { parserSettings.maxContentLength }.isSuccess, "The provided ParserSettings is a generic object that does not contain the client-specific settings.") @@ -86,6 +92,11 @@ private[pekko] object ClientConnectionSettingsImpl socketOptions = SocketOptionSettings.fromSubConfig(root, c.getConfig("socket-options")), parserSettings = ParserSettingsImpl.fromSubConfig(root, c.getConfig("parsing")), streamCancellationDelay = c.getFiniteDuration("stream-cancellation-delay"), + strictResponseEntityTimeout = c.getString("strict-response-entity-timeout").toRootLowerCase match { + case "off" => None + case _ => Some(c.getFiniteDuration("strict-response-entity-timeout")) + }, + strictResponseEntityMaxBytes = c.getPossiblyInfiniteBytes("strict-response-entity-max-bytes"), localAddress = None, http2Settings = Http2ClientSettingsImpl.fromSubConfig(root, c.getConfig("http2")), transport = ClientTransport.TCP) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala index 85479451c..d12346beb 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala @@ -17,7 +17,7 @@ import org.apache.pekko import pekko.NotUsed import pekko.actor.Cancellable import pekko.annotation.InternalApi -import pekko.http.scaladsl.model.HttpEntity +import pekko.http.scaladsl.model.{ HttpEntity, HttpResponse } import pekko.http.scaladsl.util.FastFuture import pekko.stream._ import pekko.stream.impl.fusing.GraphInterpreter @@ -273,6 +273,33 @@ private[http] object StreamUtils { def statefulAttrsMap[T, U](functionConstructor: Attributes => T => U): Flow[T, U, NotUsed] = Flow[T].via(ExposeAttributes[T, U](functionConstructor)) + /** + * Collects response entities into `HttpEntity.Strict` entities using the given timeout and maximum number of bytes. + * Responses that already carry a strict entity are passed through unchanged. + * + * With `parallelism == 1` responses keep their original order, which is what the HTTP/1.1 client needs where + * responses on a connection are sequential anyway. With a bigger value up to `parallelism` entities are collected + * concurrently and responses are emitted in the order in which their entities complete. + */ + def strictifyResponseEntities( + timeout: FiniteDuration, + maxBytes: Long, + parallelism: Int): Flow[HttpResponse, HttpResponse, NotUsed] = { + def strictify(response: HttpResponse): Source[HttpResponse, Any] = + response.entity match { + case _: HttpEntity.Strict => Source.single(response) + case entity => + entity.dataBytes + .via(new ToStrict(timeout, Some(maxBytes), entity.contentType)) + .map(strict => response.withEntity(strict)) + } + + val flow = + if (parallelism <= 1) Flow[HttpResponse].flatMapConcat(strictify) + else Flow[HttpResponse].flatMapMerge(parallelism, strictify) + flow.named("strictifyResponseEntities") + } + trait ScheduleSupport extends GraphStageLogic { self => /** diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala index fcb567be8..2976077c8 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala @@ -64,6 +64,21 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti */ final def getStreamCancellationDelay: JDuration = streamCancellationDelay.toJava final def getRequestHeaderSizeHint: Int = requestHeaderSizeHint + + /** + * If present, response entities are collected into strict entities (using the given timeout) + * before the response is dispatched to the application. Only supported by the HTTP/1.1 client. + * + * @since 2.0.0 + */ + final def getStrictResponseEntityTimeout: Optional[JDuration] = strictResponseEntityTimeout.map(_.toJava).toJava + + /** + * The maximum number of bytes collected per response entity when a strict response entity timeout is set. + * + * @since 2.0.0 + */ + final def getStrictResponseEntityMaxBytes: Long = strictResponseEntityMaxBytes final def getWebsocketSettings: WebSocketSettings = websocketSettings final def getWebsocketRandomFactory: Supplier[Random] = () => websocketRandomFactory() final def getLocalAddress: Optional[InetSocketAddress] = localAddress.toJava @@ -79,6 +94,9 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti def withRequestHeaderSizeHint(newValue: Int): ClientConnectionSettings def withStreamCancellationDelay(newValue: FiniteDuration): ClientConnectionSettings + /** @since 2.0.0 */ + def withStrictResponseEntityMaxBytes(newValue: Long): ClientConnectionSettings + // Java API versions of mutators /** @@ -114,6 +132,13 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti def withLocalAddress(newValue: Optional[InetSocketAddress]): ClientConnectionSettings = self.copy(localAddress = newValue.toScala) + /** + * Java API + * @since 2.0.0 + */ + def withStrictResponseEntityTimeout(newValue: Optional[JDuration]): ClientConnectionSettings = + self.copy(strictResponseEntityTimeout = newValue.toScala.map(_.toScala)) + @ApiMayChange def withTransport(newValue: ClientTransport): ClientConnectionSettings = self.copy(transport = newValue.asScala) } diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala index 69c99aa3b..0b7960c09 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala @@ -47,6 +47,21 @@ abstract class ClientConnectionSettings private[pekko] () def parserSettings: ParserSettings def logUnencryptedNetworkBytes: Option[Int] def streamCancellationDelay: FiniteDuration + + /** + * If defined, response entities are collected into `HttpEntity.Strict` entities (using the given timeout) + * before the response is dispatched to the application. Only supported by the HTTP/1.1 client. + * + * @since 2.0.0 + */ + def strictResponseEntityTimeout: Option[FiniteDuration] + + /** + * The maximum number of bytes collected per response entity when [[strictResponseEntityTimeout]] is defined. + * + * @since 2.0.0 + */ + def strictResponseEntityMaxBytes: Long def localAddress: Option[InetSocketAddress] def http2Settings: Http2ClientSettings @@ -64,6 +79,14 @@ abstract class ClientConnectionSettings private[pekko] () def withStreamCancellationDelay(newValue: FiniteDuration): ClientConnectionSettings = self.copy(streamCancellationDelay = newValue) + /** @since 2.0.0 */ + def withStrictResponseEntityTimeout(newValue: Option[FiniteDuration]): ClientConnectionSettings = + self.copy(strictResponseEntityTimeout = newValue) + + /** @since 2.0.0 */ + override def withStrictResponseEntityMaxBytes(newValue: Long): ClientConnectionSettings = + self.copy(strictResponseEntityMaxBytes = newValue) + // overloads for idiomatic Scala use def withWebsocketSettings(newValue: WebSocketSettings): ClientConnectionSettings = self.copy(websocketSettings = newValue) diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala index a23cd996b..d309f1e3d 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala +++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala @@ -105,6 +105,26 @@ class HttpConfigurationSpec extends PekkoSpec { } } + "have strict response entities disabled by default" in { + ClientConnectionSettings(system).strictResponseEntityTimeout should ===(None) + ClientConnectionSettings(system).strictResponseEntityMaxBytes should ===(8L * 1024 * 1024) + } + + "set `pekko.http.client.strict-response-entity-timeout`" in { + configuredSystem("""pekko.http.client.strict-response-entity-timeout = 5s + |pekko.http.client.strict-response-entity-max-bytes = 1m""".stripMargin) { sys => + import scala.concurrent.duration._ + + val client = ClientConnectionSettings(sys) + client.strictResponseEntityTimeout should ===(Some(5.seconds)) + client.strictResponseEntityMaxBytes should ===(1024L * 1024) + + val pool = ConnectionPoolSettings(sys) + pool.connectionSettings.strictResponseEntityTimeout should ===(Some(5.seconds)) + pool.connectionSettings.strictResponseEntityMaxBytes should ===(1024L * 1024) + } + } + "change parser settings for all by setting `pekko.http.parsing`" in { configuredSystem("""pekko.http.parsing.illegal-header-warnings = off""") { sys => val client = ClientConnectionSettings(sys) diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala index fe48e6205..ad0cd246d 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala +++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala @@ -1003,6 +1003,121 @@ class LowLevelOutgoingConnectionSpec extends PekkoSpecWithMaterializer with Insi netInSub.sendComplete() responses.expectComplete() } + + "collect response entities into strict entities if configured".which { + val strictConfig = "pekko.http.client.strict-response-entity-timeout = 2s" + + "collects a chunked response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |4 + |DEFG + |0 + | + |""") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "collects a default response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Content-Length: 7 + | + |ABC""") + sendWireData("DEFG") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "collects a close-delimited response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + | + |ABCDEFG""") + closeNetworkInput() + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + } + + "passes through an already strict response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Content-Length: 7 + | + |ABCDEFG""") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "fails the connection if the response entity exceeds strict-response-entity-max-bytes" in new TestSetup( + config = strictConfig + "\npekko.http.client.strict-response-entity-max-bytes = 5") { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |4 + |DEFG + |0 + | + |""") + + responsesSub.request(1) + responses.expectError().getMessage should include("longer than the maximum of 5") + } + + "leaves response entities streamed by default" in new TestSetup { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |""") + + inside(expectResponse()) { + case HttpResponse(_, _, entity: HttpEntity.Chunked, _) => + entity.contentType shouldEqual ContentTypes.`text/plain(UTF-8)` + } + } + } } class TestSetup(maxResponseContentLength: Int = -1, config: String = "") { diff --git a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala index d17c0185f..82e334678 100644 --- a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala +++ b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala @@ -717,6 +717,85 @@ class Http2ClientSpec extends PekkoSpecWithMaterializer(""" } + "support strict response entities" should { + abstract class StrictEntitySetup extends TestSetup with NetProbes { + override def settings = + super.settings.withStrictResponseEntityTimeout(Some(2.seconds.dilated)) + } + + "collect a response entity into a strict entity".inAssertAllStagesStopped(new StrictEntitySetup { + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(streamId, endStream = false, ByteString("abc")) + network.sendDATA(streamId, endStream = true, ByteString("def")) + + user.expectResponse().entity shouldBe + HttpEntity.Strict(ContentTypes.`application/octet-stream`, ByteString("abcdef")) + }) + + "emit responses in the order in which their entities complete".inAssertAllStagesStopped( + new StrictEntitySetup { + user.emitRequest(Get("/slow")) + network.expectDecodedHEADERS(0x1, endStream = true) + user.emitRequest(Get("/fast")) + network.expectDecodedHEADERS(0x3, endStream = true) + + // the first response starts first but its entity stays incomplete + network.sendHEADERS(0x1, endStream = false, + Seq(RawHeader(":status", "200"), RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(0x1, endStream = false, ByteString("in")) + + // while the second response completes right away + network.sendHEADERS(0x3, endStream = false, + Seq(RawHeader(":status", "201"), RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(0x3, endStream = true, ByteString("complete")) + + user.expectResponse().status shouldBe StatusCodes.Created + + network.sendDATA(0x1, endStream = true, ByteString("complete")) + user.expectResponse().status shouldBe StatusCodes.OK + }) + + "fail the connection if a response entity exceeds strict-response-entity-max-bytes".inAssertAllStagesStopped( + new StrictEntitySetup { + override def settings = super.settings.withStrictResponseEntityMaxBytes(5) + + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + EventFilter.error(pattern = "HTTP2 connection failed with error .*", occurrences = 1).intercept { + network.sendDATA(streamId, endStream = true, ByteString("abcdef")) + + user.responseIn.expectSubscriptionAndError().getMessage should include( + "longer than the maximum of 5") + } + }) + + "leave response entities streamed by default".inAssertAllStagesStopped(new TestSetup with NetProbes { + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + + user.expectResponse().entity shouldBe a[Chunked] + }) + } + "expose synthetic headers" should { "expose Tls-Session-Info".inAssertAllStagesStopped(new TestSetup { lazy val expectedSession = SSLContext.getDefault.createSSLEngine.getSession