Fix MultipartParser & PartGenerator memory leak

Prior to this commit, the reactive `MultipartParser` and `PartGenerator`
types were leaking memory at runtime in specific cases:

* many HTTP clients must send multipart requests to be parsed and close
  the connection while uploading
* the `PartGenerator` must be configured to write file parts to
  temporary files on disk
* concurrency, upload speed must be important to trigger cases where the
  file system is not fast enough to consume incoming buffers

The `MultipartParser` parses and emits `BodyToken` to its sink
(here, the `PartGenerator`). By definition, Reactor's `FluxSink` when
created with `Flux.create(FluxSink)` will use a "buffer" strategy and
will queue emitted elements if they cannot be consumed.

Here, the cancellation signal does dispose internal states in the
`MultiPartParser` and `PartGenerator` but does not clear the internal
queue in `FluxSink`.

This commit ensures that an operation is registered to release buffers
on the discard event.

Fixes gh-36262
This commit is contained in:
Brian Clozel
2026-02-13 18:45:05 +01:00
parent 50bffe7ddc
commit 474d520182
@@ -99,7 +99,8 @@ final class PartGenerator extends BaseSubscriber<MultipartParser.Token> {
sink.onCancel(generator);
sink.onRequest(l -> generator.requestToken());
tokens.subscribe(generator);
tokens.doOnDiscard(MultipartParser.BodyToken.class, bodyToken -> DataBufferUtils.release(bodyToken.buffer()))
.subscribe(generator);
});
}