Design a file downloader library.

package com.androidinterview.filedownloader;

import java.io.File;

// Built once and never mutated, so a queued download cannot change under the
// worker that is already running it. The id is what survives process death,
// because the resume offset has to be filed under something stable.
public record DownloadRequest(String id, String url, File directory, String fileName, int priority) {

    public File temporaryFile() {
        return new File(directory, fileName + ".part");
    }

    public File finishedFile() {
        return new File(directory, fileName);
    }
}

package com.androidinterview.filedownloader;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.net.HttpURLConnection;
import java.net.URI;

// One download, resumable, cancellable, and reporting progress on an interval.
// Everything in this class exists because the file is large enough that
// starting over is expensive. On Android this runs inside a foreground
// service or a WorkManager worker, which is what keeps the process alive for
// a transfer that outlives the screen that started it.
public final class DownloadTask implements Runnable, Comparable<DownloadTask> {

    public interface Listener {
        // totalBytes is -1 when the server did not say, a chunked response.
        void onProgress(long bytesWritten, long totalBytes);

        void onComplete(File file);

        // Its own outcome, not a failure. A UI shows paused for this and
        // retry for onFailed, and it cannot tell them apart from an exception.
        void onCancelled(long bytesWritten);

        void onFailed(Exception cause);
    }

    private static final int CHUNK = 8 * 1024;
    private static final long REPORT_INTERVAL_MILLIS = 200;
    // Generous on read, because a slow transfer that is still moving is not a
    // failure. In a real build these live on the OkHttp client and Call.cancel
    // replaces the flag below.
    private static final int CONNECT_TIMEOUT_MILLIS = 15_000;
    private static final int READ_TIMEOUT_MILLIS = 60_000;

    private final DownloadRequest request;
    private final ProgressStore store;
    private final Listener listener;
    private final Runnable onTerminated;
    private volatile boolean cancelled;

    public DownloadTask(DownloadRequest request, ProgressStore store, Listener listener, Runnable onTerminated) {
        this.request = request;
        this.store = store;
        this.listener = listener;
        this.onTerminated = onTerminated;
    }

    public void cancel() {
        cancelled = true;
    }

    @Override
    public void run() {
        try {
            if (cancelled) {
                // Cancelled while still queued. Nothing was opened, so the
                // resume point is whatever the last run persisted.
                listener.onCancelled(resumeOffset());
                return;
            }
            long written = transfer(resumeOffset());
            if (cancelled) {
                // Pause and cancel are one code path. The offset is persisted
                // either way, so resuming is the same request with a new Range
                // header and no special case.
                listener.onCancelled(written);
                return;
            }
            File finished = request.finishedFile();
            // Rename last. A file appears at its real path only once it is
            // whole, so a process killed mid transfer can never leave
            // something a later reader would treat as a complete download.
            if (!request.temporaryFile().renameTo(finished)) {
                throw new IOException("Could not commit " + request.temporaryFile());
            }
            store.clear(request.id());
            listener.onProgress(written, written);
            listener.onComplete(finished);
        } catch (Exception failure) {
            listener.onFailed(failure);
        } finally {
            onTerminated.run();
        }
    }

    // The persisted offset is written on a throttle, so after a kill the part
    // file is usually a little longer than the saved offset, which is safe.
    // The other way round would mean seeking past the end and leaving a hole
    // of zeros in the middle of the file, so the file's length is the ceiling.
    private long resumeOffset() {
        if (!request.temporaryFile().exists()) {
            return 0;
        }
        return Math.min(store.bytesWritten(request.id()), request.temporaryFile().length());
    }

    private long transfer(long resumeFrom) throws IOException {
        HttpURLConnection connection = (HttpURLConnection) URI.create(request.url()).toURL().openConnection();
        connection.setConnectTimeout(CONNECT_TIMEOUT_MILLIS);
        connection.setReadTimeout(READ_TIMEOUT_MILLIS);
        if (resumeFrom > 0) {
            connection.setRequestProperty("Range", "bytes=" + resumeFrom + "-");
        }
        connection.connect();

        // A server that does not honour Range answers 200 with the whole file
        // rather than 206 with the tail. Any bytes already on disk are then
        // meaningless, so the offset resets and the download starts over. This
        // fallback is not optional, plenty of servers ignore Range.
        long start = connection.getResponseCode() == HttpURLConnection.HTTP_PARTIAL ? resumeFrom : 0;
        // A chunked response has no Content-Length and reports -1. Adding an
        // offset to -1 would make the total smaller than the offset, so
        // unknown stays unknown and the length check below is skipped.
        long length = connection.getContentLengthLong();
        long total = length < 0 ? -1 : start + length;

        long written = start;
        long lastReport = 0;
        try (InputStream source = connection.getInputStream();
             RandomAccessFile sink = new RandomAccessFile(request.temporaryFile(), "rw")) {
            sink.seek(start);
            // Truncate to the start point. On the 200 fallback that is zero,
            // and without it a fresh body shorter than the stale bytes already
            // on disk would leave the old tail in place, pass the length check
            // and be renamed into place corrupt.
            sink.setLength(start);
            if (start == 0) {
                store.saveProgress(request.id(), 0);
            }
            byte[] buffer = new byte[CHUNK];
            for (int read; (read = source.read(buffer)) != -1; ) {
                if (cancelled) {
                    break;
                }
                sink.write(buffer, 0, read);
                written += read;
                long now = System.currentTimeMillis();
                if (now - lastReport >= REPORT_INTERVAL_MILLIS) {
                    // Both of these are throttled together. A callback per
                    // chunk floods the main thread with updates no eye can
                    // resolve, and a database write per chunk costs more I O
                    // than the download itself.
                    listener.onProgress(written, total);
                    store.saveProgress(request.id(), written);
                    lastReport = now;
                }
            }
        } finally {
            // Whatever ended the loop, cancel, failure or completion, the
            // resume point is exact rather than up to 200 milliseconds stale.
            store.saveProgress(request.id(), written);
        }
        if (cancelled) {
            return written;
        }
        // The length check is the cheap integrity check. When the server sends
        // a digest header, verify that instead, a truncated file that happens
        // to have the right length is still corrupt.
        if (total >= 0 && written != total) {
            throw new IOException("Truncated at " + written + " of " + total);
        }
        return written;
    }

    // Higher priority first, so a download the user is watching preempts a
    // background prefetch instead of queueing behind it.
    @Override
    public int compareTo(DownloadTask other) {
        return Integer.compare(other.request.priority(), request.priority());
    }
}

package com.androidinterview.filedownloader;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

// The queue in front of the tasks. Downloads are I O bound, so the pool can
// run more of them than there are cores, but it still needs a cap. Dozens of
// parallel connections to one host get you throttled rather than served
// faster, and on a thin connection they only split the same bandwidth.
public final class Downloader {

    private final ThreadPoolExecutor pool;
    private final ProgressStore store;
    private final Map<String, DownloadTask> running = new ConcurrentHashMap<>();

    public Downloader(int concurrency, ProgressStore store) {
        this.store = store;
        // The tasks are queued directly rather than submitted, because submit
        // wraps a runnable in a FutureTask and the priority queue would then
        // be ordering wrappers that cannot compare themselves.
        this.pool = new ThreadPoolExecutor(
                concurrency, concurrency, 0, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>());
    }

    public void enqueue(DownloadRequest request, DownloadTask.Listener listener) {
        DownloadTask task =
                new DownloadTask(request, store, listener, () -> running.remove(request.id()));
        running.put(request.id(), task);
        pool.execute(task);
    }

    // Cancelling a queued download and cancelling a running one are the same
    // call. The offset is already persisted, so enqueueing the same id later
    // resumes rather than restarts. A task pulled out of the queue before it
    // ran is run inline, so the listener still hears cancelled. It opens no
    // connection, because the flag is checked first.
    public void cancel(String id) {
        DownloadTask task = running.remove(id);
        if (task != null) {
            task.cancel();
            if (pool.remove(task)) {
                task.run();
            }
        }
    }
}

package com.androidinterview.filedownloader;

// Where the resume offset lives, a Room table in a real app. It is on disk
// rather than in memory because the interesting case is not a dropped
// connection, it is the process being killed while a two hundred megabyte
// download was half done.
public interface ProgressStore {

    long bytesWritten(String id);

    void saveProgress(String id, long bytes);

    void clear(String id);
}

package com.androidinterview.filedownloader

import java.io.File

// Built once and never mutated, so a queued download cannot change under the
// worker already running it. The id is what survives process death, because
// the resume offset has to be filed under something stable.
data class DownloadRequest(
    val id: String,
    val url: String,
    val directory: File,
    val fileName: String,
    val priority: Int = 0,
) {
    val temporaryFile get() = File(directory, "$fileName.part")
    val finishedFile get() = File(directory, fileName)
}

// Where the resume offset lives, a Room table in a real app. It is on disk
// rather than in memory because the interesting case is not a dropped
// connection, it is the process being killed while a large download was half
// done.
interface ProgressStore {
    fun bytesWritten(id: String): Long
    fun saveProgress(id: String, bytes: Long)
    fun clear(id: String)
}

sealed interface DownloadEvent {
    // totalBytes is -1 when the server did not say, a chunked response.
    data class Progress(val bytesWritten: Long, val totalBytes: Long) : DownloadEvent
    data class Complete(val file: File) : DownloadEvent

    // Its own outcome, not a failure. A UI shows paused for this and retry
    // for Failed, and it cannot tell them apart from an exception.
    data class Cancelled(val bytesWritten: Long) : DownloadEvent
    data class Failed(val cause: Exception) : DownloadEvent
}

package com.androidinterview.filedownloader

import java.io.IOException
import java.io.RandomAccessFile
import java.net.HttpURLConnection
import java.net.URI

private const val CHUNK = 8 * 1024
private const val REPORT_INTERVAL_MILLIS = 200L

// Generous on read, because a slow transfer that is still moving is not a
// failure. In a real build these live on the OkHttp client and Call.cancel
// replaces the flag below.
private const val CONNECT_TIMEOUT_MILLIS = 15_000
private const val READ_TIMEOUT_MILLIS = 60_000

// One download, resumable, cancellable, and reporting progress on an interval.
// Everything here exists because the file is large enough that starting over
// is expensive. On Android this runs inside a foreground service or a
// WorkManager worker, which is what keeps the process alive for a transfer
// that outlives the screen that started it.
class DownloadTask(
    private val request: DownloadRequest,
    private val store: ProgressStore,
    private val onEvent: (DownloadEvent) -> Unit,
    private val onTerminated: () -> Unit = {},
) : Runnable, Comparable<DownloadTask> {

    @Volatile
    private var cancelled = false

    fun cancel() {
        cancelled = true
    }

    override fun run() {
        try {
            if (cancelled) {
                // Cancelled while still queued. Nothing was opened, so the
                // resume point is whatever the last run persisted.
                onEvent(DownloadEvent.Cancelled(resumeOffset()))
                return
            }
            val written = transfer(resumeOffset())
            if (cancelled) {
                // Pause and cancel are one code path. The offset is persisted
                // either way, so resuming is the same request with a new Range
                // header and no special case.
                onEvent(DownloadEvent.Cancelled(written))
                return
            }
            // Rename last. A file appears at its real path only once it is
            // whole, so a process killed mid transfer can never leave
            // something a later reader treats as a finished download.
            check(request.temporaryFile.renameTo(request.finishedFile)) { "Could not commit ${request.temporaryFile}" }
            store.clear(request.id)
            onEvent(DownloadEvent.Progress(written, written))
            onEvent(DownloadEvent.Complete(request.finishedFile))
        } catch (failure: Exception) {
            onEvent(DownloadEvent.Failed(failure))
        } finally {
            onTerminated()
        }
    }

    // The persisted offset is written on a throttle, so after a kill the part
    // file is usually a little longer than the saved offset, which is safe.
    // The other way round would mean seeking past the end and leaving a hole
    // of zeros in the middle of the file, so the file's length is the ceiling.
    private fun resumeOffset(): Long =
        if (request.temporaryFile.exists()) minOf(store.bytesWritten(request.id), request.temporaryFile.length()) else 0

    private fun transfer(resumeFrom: Long): Long {
        val connection = URI.create(request.url).toURL().openConnection() as HttpURLConnection
        connection.connectTimeout = CONNECT_TIMEOUT_MILLIS
        connection.readTimeout = READ_TIMEOUT_MILLIS
        if (resumeFrom > 0) connection.setRequestProperty("Range", "bytes=$resumeFrom-")
        connection.connect()

        // A server that does not honour Range answers 200 with the whole file
        // rather than 206 with the tail. Bytes already on disk are then
        // meaningless, so the offset resets and the download starts over. This
        // fallback is not optional, plenty of servers ignore Range.
        val start = if (connection.responseCode == HttpURLConnection.HTTP_PARTIAL) resumeFrom else 0L
        // A chunked response has no Content-Length and reports -1. Adding an
        // offset to -1 would make the total smaller than the offset, so
        // unknown stays unknown and the length check below is skipped.
        val total = connection.contentLengthLong.let { if (it < 0) -1L else start + it }

        var written = start
        var lastReport = 0L
        try {
            connection.inputStream.use { source ->
                RandomAccessFile(request.temporaryFile, "rw").use { sink ->
                    sink.seek(start)
                    // Truncate to the start point. On the 200 fallback that is
                    // zero, and without it a fresh body shorter than the stale
                    // bytes already on disk would leave the old tail in place,
                    // pass the length check and be renamed into place corrupt.
                    sink.setLength(start)
                    if (start == 0L) store.saveProgress(request.id, 0)
                    val buffer = ByteArray(CHUNK)
                    while (!cancelled) {
                        val read = source.read(buffer)
                        if (read < 0) break
                        sink.write(buffer, 0, read)
                        written += read
                        val now = System.currentTimeMillis()
                        if (now - lastReport >= REPORT_INTERVAL_MILLIS) {
                            // Throttled together. A callback per chunk floods
                            // the main thread with updates no eye can resolve,
                            // and a database write per chunk costs more than
                            // the download itself.
                            onEvent(DownloadEvent.Progress(written, total))
                            store.saveProgress(request.id, written)
                            lastReport = now
                        }
                    }
                }
            }
        } finally {
            // Whatever ended the loop, cancel, failure or completion, the
            // resume point is exact rather than up to 200 milliseconds stale.
            store.saveProgress(request.id, written)
        }
        if (cancelled) return written
        // The length check is the cheap integrity check. When the server sends
        // a digest header, verify that instead, a truncated file that happens
        // to have the right length is still corrupt.
        if (total >= 0 && written != total) throw IOException("Truncated at $written of $total")
        return written
    }

    // Higher priority first, so a download the user is watching preempts a
    // background prefetch instead of queueing behind it.
    override fun compareTo(other: DownloadTask) = other.request.priority.compareTo(request.priority)
}

package com.androidinterview.filedownloader

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.PriorityBlockingQueue
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

// The queue in front of the tasks. Downloads are I O bound, so the pool can
// run more of them than there are cores, but it still needs a cap. Dozens of
// parallel connections to one host get you throttled rather than served
// faster, and on a thin connection they only split the same bandwidth.
class Downloader(concurrency: Int, private val store: ProgressStore) {

    // Tasks are queued rather than submitted, because submit wraps a runnable
    // in a FutureTask and the priority queue would then be ordering wrappers
    // that cannot compare themselves.
    private val pool = ThreadPoolExecutor(
        concurrency, concurrency, 0, TimeUnit.MILLISECONDS, PriorityBlockingQueue(),
    )

    private val running = ConcurrentHashMap<String, DownloadTask>()

    fun enqueue(request: DownloadRequest, onEvent: (DownloadEvent) -> Unit) {
        val task = DownloadTask(request, store, onEvent) { running.remove(request.id) }
        running[request.id] = task
        pool.execute(task)
    }

    // Cancelling a queued download and cancelling a running one are the same
    // call. The offset is already persisted, so enqueueing the same id later
    // resumes rather than restarts. A task pulled out of the queue before it
    // ran is run inline, so the listener still hears cancelled. It opens no
    // connection, because the flag is checked first.
    fun cancel(id: String) {
        val task = running.remove(id) ?: return
        task.cancel()
        if (pool.remove(task)) task.run()
    }
}