Ensure deflater/inflater end() is called in DeflateCompressor and GzipCompressor#1142
Ensure deflater/inflater end() is called in DeflateCompressor and GzipCompressor#1142pjfanning wants to merge 9 commits into
Conversation
pjfanning
changed the title
…pCompressor - DeflateCompressor: add deflaterEnded flag + idempotent endDeflater() method; call endDeflater() in finishWithBuffer() instead of deflater.end() directly - GzipDecompressor: add postStop() calling inflater.end() + add createInflater() factory method for testability - StreamUtils.byteStringTransformer: add optional cleanup callback, call it in postStop() when the stage is stopped before onUpstreamFinish() - Encoder.singleUseEncoderFlow: pass cleanup callback that calls endDeflater() on DeflateCompressor instances to handle stream cancellation/failure - DeflateSpec: add tests for deflater cleanup on normal finish and cancellation - GzipSpec: add tests for inflater cleanup on normal finish and cancellation, and deflater cleanup on normal finish and cancellation
pjfanning
force-pushed
the
copilot/ensure-end-called-on-inflaters-deflaters
branch
from
7653520 to
8add717
Compare
He-Pin
left a comment
He-Pin
left a comment
There was a problem hiding this comment.
Overall this looks like a solid cleanup. The idempotent endDeflater() pattern is clean, and the tests actually exercise the cancellation path properly (nice touch with awaitEnd to avoid the postStop race).
A couple of minor thoughts inline. Also — the PR title still has "WIP" in it, might want to update that before merging.
pjfanning
changed the title
| if (data.nonEmpty) emit(out, data) | ||
| completeStage() | ||
| } finally { | ||
| finished = true |
There was a problem hiding this comment.
Setting the finished flag in the finally block prevents postStop() from calling cleanup() if finish() throws. In that failure path the stage still stops, but the finished guard is already true, so the deflater leak this PR is trying to fix can still happen. Please move the flag assignment onto the success path, right before completeStage(). That keeps the happy path idempotent while still letting postStop() clean up on exceptions.
There was a problem hiding this comment.
I moved it before the if (data.nonEmpty) emit(out, data) - is that ok?
| } | ||
| val data = finish() | ||
| finished = true | ||
| if (data.nonEmpty) emit(out, data) |
There was a problem hiding this comment.
then we should add the finished = true to the emit(out, data, () => finished)` wdyt
more changes related to #1133