Understanding DataBufferUtils.write to OutputStream

I'm downloading huge documents from internal servers in a kind of proxy. In a nutshell, I have a Flux<DataBuffer> and an OutputStream. So I thought I could use

org.springframework.core.io.buffer.DataBufferUtils#write(
    org.reactivestreams.Publisher<org.springframework.core.io.buffer.DataBuffer>,
    java.io.OutputStream
)

but that returns a Flux<DataBuffer> and I'm not sure what to do with it.

Flux.blockFirst() will just process the start of the file.

Flux.blockLast() logs thousands of resource leaks. I think that makes sense because the specific write() method clearly states that the caller is responsible for releasing the DataBuffers.

Flux.subscribe(DataBufferUtils.releaseConsumer()) doesn't do anything when I run the code from a unit test. My guess is that the VM terminates before Spring has a chance to start the processing.

The code from this answer works but I don't understand why:

Mono.create(sink ->
    DataBufferUtils
        .write(myFlux, outputStream)
        .subscribe(
            DataBufferUtils::release,
            sink::error,
            sink::success
        )
)
    .block()

Why does this download the document without resource leaks? How does wrapping the Flux in a Mono make a difference?

That piece of code is capable of writing given Flux and releasing resources:

DataBufferUtils
        .write(myFlux, outputStream)
        .subscribe(
            DataBufferUtils::release,

write(myFlux, outputStream) do the writing and DataBufferUtils::release is provided as a subscribe's first argument, which is a function that reactor will call on each buffer. So that's how you avoid memory leak.

Rest of code including wrapping to Mono is just an approach to block until your flux is finished, i.e. written:

Mono.create(sink ->
    // ...
            sink::error,
            sink::success
        )
)
    .block()

sink::error and sink::success is 2nd and 3rd arguments of subscribe which are functions that reactor will call when original Flux is finished (normally or exceptionally), i.e. written. Since sink here is Mono's sink calling them finishes the manually created Mono. That's how you turn a Flux to a Mono, which is completed when the Flux is completed.

Finally, calling .block on that Mono blocks a thread until Mono's (and hence the original Flux) is finished. That is were the actual block happens.

I expect here is a simpler solution:

DataBufferUtils.write(myFlux, outputStream).doOnNext(DataBufferUtils::release).blockLast()

You don't actually need extra Mono, all you need is to call DataBufferUtils::release on each buffer after is was written and block until end of Flux, i.e. blockLast. You can replace doOnNext (adds a side-effect) with map (converts items with a possible side-effect), what will turn Flux<DataBuffer> to a Flux<Boolean>, since release returns boolean. Inasmuch as you don't use neither buffer's nor booleans and just block until the end, both methods are suitable for you here.