public class Oom extends ServerHttpResponseDecorator {
    
    public Oom(ServerHttpResponse delegate) {
        super(delegate);
    }

    @Override
    @NonNull
    public Mono<Void> writeWith(@NonNull Publisher<? extends DataBuffer> body) {
        Flux<? extends DataBuffer> flux = Flux.from(body);
        return super.writeWith(flux.buffer().map(dataBuffers -> {
            DataBufferFactory dataBufferFactory = getDelegate().bufferFactory();
            DataBuffer join = dataBufferFactory.join(dataBuffers);
            byte[] content = new byte[join.readableByteCount()];
            join.read(content);
            DataBufferUtils.release(join);
            // Игнорировать другую обработку
            return dataBufferFactory.wrap(content);
        }));
    }
}

reactor.core.publisher.FluxBuffer.BufferExactSubscriber#onNext

@Override
public void onNext(T t) {
    if (done) {
        Operators.onNextDropped(t, actual.currentContext());
        return;
    }

    C b = buffer;
    if (b == null) {
        try {
            b = Objects.requireNonNull(bufferSupplier.get(),
                    "The bufferSupplier returned a null buffer");
        }
        catch (Throwable e) {
            Context ctx = actual.currentContext();
            onError(Operators.onOperatorError(s, e, t, ctx));
            Operators.onDiscard(t, ctx); //this is in no buffer
            return;
        }
        buffer = b;
    }
    // b — это список, и потребленный t будет добавлен к b
    b.add(t);

    if (b.size() == size) {
        buffer = null;
        actual.onNext(b);
    }
}

reactor.core.publisher.FluxBuffer.BufferExactSubscriber#onError

@Override
public void onError(Throwable t) {
    if (done) {
        Operators.onErrorDropped(t, actual.currentContext());
        return;
    }
    done = true;
    actual.onError(t);
    Operators.onDiscardMultiple(buffer, actual.currentContext());
}

@Override
public void cancel() {
    s.cancel();
    Operators.onDiscardMultiple(buffer, this.ctx);
}

@Override
@NonNull
public Mono<Void> writeWith(@NonNull Publisher<? extends DataBuffer> body) {
    Flux<? extends DataBuffer> flux = Flux.from(body);
    return super.writeWith(flux.buffer().map(dataBuffers -> {
        DataBufferFactory dataBufferFactory = getDelegate().bufferFactory();
        DataBuffer join = dataBufferFactory.join(dataBuffers);
        byte[] content = new byte[join.readableByteCount()];
        join.read(content);
        DataBufferUtils.release(join);
        // Игнорировать другую обработку
        return dataBufferFactory.wrap(content);
    })
    // Отслеживайте события сброса и перезапускайте буфер данных.
    .doOnDiscard(DataBuffer.class, DataBufferUtils::release));
}