[FIXED] How to track progress and speed of DataBufferUtils.write()?

Issue

I'm trying to download a large file via Spring Boot's WebClient (per advice from https://stackoverflow.com/a/60725206):

Flux<DataBuffer> dataBufferFlux = webClientEmbed.get()
  .uri(someUri)
  // ...
  .retrieve()
  .bodyToFlux(DataBuffer.class);


DataBufferUtils
  .write(dataBufferFlux, somePath, CREATE)
  .block();

The problem is, I have no idea how to track the progress and speed of the download or even detect when the write breaks (other than using doOnError which doesn't seem to be detecting every error...). I've scoured the docs and can't find any info on how to do that at all. I suspect I may be too much a of a WebFlux newb but I'm at my wit's end here.

What is the correct way to track the progress of a WebClient download when using the DataBuffer class?

Solution

here a code snippet that shows how to track progress of WebClient and DataBufferUtils.write

    public DownloadProgress download(String url, String targetFile) {

        WebClient client = WebClient.builder()
                .clientConnector(
                    new ReactorClientHttpConnector(HttpClient.create().followRedirect(true)
                 ))
                .baseUrl(url)
                .build();
        
        final DownloadProgress progress = new DownloadProgress();

        Flux<DataBuffer> dataBufferFlux = client.get()
            .exchangeToFlux( response -> {
                long contentLength = response.headers().contentLength().orElse(-1);
                progress.setLength(contentLength);
                return response.bodyToFlux(DataBuffer.class);
            });
        
        Path path = Paths.get(targetFile);
        FileOutputStream fout = null;
        try {
            fout = new FileOutputStream(path.toFile());
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Unable to create targetFile", e);
        }
        
        DataBufferUtils.write(dataBufferFlux, progress.getOutputStream(fout) )
            .subscribe( DataBufferUtils.releaseConsumer() );
        
        return progress;
    }

This method returns a "DownloadProgress" object updated while the outpout stream is written. Here is this DownloadProgress class :

public class DownloadProgress {

    private long contentLength = -1;
    private long downloaded;

    public void setLength(long contentLength) {
        this.contentLength = contentLength;
    }
    
    public OutputStream getOutputStream(FileOutputStream fileOutputStream) {
        return new FilterOutputStream(fileOutputStream) {
            @Override
            public void write(byte[] b, int off, int len) throws IOException {
                out.write(b, off, len);
                downloaded += len;
            }
            @Override
            public void write(int b) throws IOException {
                out.write(b);
                downloaded++;
            }
            @Override
            public void close() throws IOException {
                super.close();
                done();
            }
        };
    }
    
    public void done() {
        if ( contentLength == -1 ) {
            contentLength = downloaded;
        } else {
            downloaded = contentLength;
        }
    }
    
    public double getProgress() {
        if ( contentLength == -1 ) return 0;
        return downloaded / (double) contentLength;
    }
    
    public long getDownloaded() {
        return downloaded;
    }

    public boolean finished() {
        return downloaded > 0 && downloaded == contentLength;
    }
    
}

Then, to monitor the download you can loop while finished returns false and display the progress.

            do {
                String progress = String.format("%.2f", dl.getProgress() * 100.0);
                log.info( "Downloaded: {}%", progress );
                try {
                    Thread.sleep(250);
                } catch (InterruptedException e) {
                    return;
                }
            } while(!dl.finished());

Hope this could help.

Answered By - Guillaume Dufrêne