He-Pin commented on code in PR #1137:
URL: https://github.com/apache/pekko-http/pull/1137#discussion_r3529586442
##########
http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala:
##########
@@ -103,14 +103,27 @@ private[coding] object DeflateCompressor {
private[coding] class DeflateDecompressor(
maxBytesPerChunk: Int = Decoder.MaxBytesPerChunkDefault) extends
DeflateDecompressorBase(maxBytesPerChunk) {
+ protected[coding] def createInflater(noWrap: Boolean): Inflater = new
Inflater(noWrap)
+
override def createLogic(attr: Attributes) = new ParsingLogic {
+ private var currentInflater: Inflater = null
+ private var inflaterEnded = false
+
+ private def cleanupInflater(): Unit =
+ if (!inflaterEnded && currentInflater != null) {
+ inflaterEnded = true
+ currentInflater.end()
+ }
+
+ override def postStop(): Unit = cleanupInflater()
/** Step that probes if the deflate stream contains a zlib wrapper */
case object ProbeWrapping extends ParseStep[ByteString] {
override def onTruncation(): Unit = completeStage()
override def parse(reader: ByteStringParser.ByteReader):
ParseResult[ByteString] = {
val inflater = examineAndBuildInflater(reader.remainingData)
+ currentInflater = inflater
Review Comment:
Can we close the previous inflater before replacing it here? loops back to
after , so concatenated deflate streams can run this path more than once. As
written, the second assignment overwrites the first inflater and can only the
last one. A small helper that calls the idempotent cleanup before assignment,
then resets the guard for the new inflater, would keep the leak fix complete.
##########
http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala:
##########
@@ -61,6 +64,54 @@ class DeflateSpec extends CoderSpec {
3.seconds.dilated)
.awaitResult(3.seconds.dilated) should equal(request)
}
+ "release the inflater when decoding completes" in {
+ val inflater = new TrackingInflater
+ decodeWith(inflater, streamEncode(smallTextBytes)) should
readAs(smallText)
+ inflater.endCalls.get() shouldEqual 1
+ }
+ "release the inflater when decoding is cancelled early" in {
+ val inflater = new TrackingInflater
+ val compressed = streamEncode(largeTextBytes)
+
+ Source.single(compressed)
+ .via(decoderWith(inflater).withMaxBytesPerChunk(1).decoderFlow)
+ .take(1)
+ .runWith(Sink.ignore)
+ .awaitResult(3.seconds.dilated)
+
+ inflater.endCalls.get() shouldEqual 1
+ }
+ "release the inflater when decoding is truncated" in {
+ val inflater = new TrackingInflater
+ // Truncated deflate streams complete without exception (completeStage
on truncation)
+ decodeWith(inflater, streamEncode(smallTextBytes).dropRight(5))
+ inflater.endCalls.get() shouldEqual 1
+ }
+ }
+
+ @nowarn("msg=deprecated")
Review Comment:
This currently breaks the Scala 2.13 CI: at this line. It looks like the
deprecated warning is only introduced by , so this annotation should be
dropped (or moved to the expression that actually needs it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
