Resumable file uploads in Angular with RxJS

Tim Koschützki

Co-founder · Berlin, Germany ·

Resume an interrupted Angular upload from the bytes the receiver has committed, even when the browser missed its last response. This walkthrough builds a single-file Angular/RxJS form, a local Node.js receiver, and the proxy connecting them. After publication succeeds, you can compare the saved file with the original.

Why chunked and resumable uploads?

An interrupted request may have reached the server even when its response never reached the browser. Resume therefore reads the server’s committed offset instead of trusting a local chunk counter. This example sends 1 MiB chunks sequentially and supports pause, resume, and cancel within the same page. Reloading loses the selected File and UUID; restarting the receiver loses its session state. Neither action can resume the previous upload here.

If you only need a single multipart request, use our basic Angular upload walkthrough. For standardized resumability that also survives page reloads, use the separate Angular tus tutorial.

Setting up your Angular project

Use Bash, Node.js 24.15.0, Corepack with Yarn 4.12.0, and a modern browser. The pins below use Angular 22.2.0, TypeScript 6.0.3, and RxJS 7.8.2. This browser-only example was also tested with Node.js 26.8.1 and Chromium 145 on Linux. Check Angular’s compatibility requirements when changing those versions.

From your chosen parent directory, paste this block. It refuses an existing destination and an enclosing Yarn Plug’n’Play project, whose loader can affect the Angular build. The subshell keeps your current directory and shell options unchanged if any step fails:

(
  node --input-type=module <<'NODE' &&
import { existsSync, realpathSync } from 'node:fs'
import { dirname, join } from 'node:path'
let directory = realpathSync('.')
while (true) {
  if (existsSync(join(directory, '.pnp.cjs')) || existsSync(join(directory, '.pnp.js'))) {
    console.error('Create this demo outside the enclosing Yarn Plug’n’Play project.')
    process.exit(1)
  }
  const parent = dirname(directory)
  if (parent === directory) break
  directory = parent
}
NODE
  mkdir chunked-upload-app &&
  cd chunked-upload-app &&
  printf 'nodeLinker: node-modules\nenableGlobalCache: false\n' > .yarnrc.yml &&
  touch yarn.lock
)

Save all the following files inside chunked-upload-app, creating src/app/ in your editor. package.json pins the direct dependencies; retain the lockfile produced by installation:

{
  "name": "chunked-upload-app",
  "private": true,
  "type": "module",
  "packageManager": "yarn@4.12.0",
  "scripts": {
    "build": "ng build",
    "start": "ng serve --host 127.0.0.1 --port 4200"
  },
  "browserslist": ["Chrome 145"],
  "dependencies": {
    "@angular/build": "22.2.0",
    "@angular/cli": "22.2.0",
    "@angular/common": "22.2.0",
    "@angular/compiler": "22.2.0",
    "@angular/compiler-cli": "22.2.0",
    "@angular/core": "22.2.0",
    "@angular/platform-browser": "22.2.0",
    "rxjs": "7.8.2",
    "tslib": "2.8.1",
    "typescript": "6.0.3"
  }
}

Save angular.json:

{
  "version": 1,
  "cli": { "analytics": false, "cache": { "enabled": false } },
  "projects": {
    "demo": {
      "projectType": "application",
      "root": ".",
      "architect": {
        "build": {
          "builder": "@angular/build:application",
          "options": {
            "browser": "src/main.ts",
            "tsConfig": "tsconfig.json",
            "index": "src/index.html",
            "outputPath": "dist",
            "styles": [],
            "assets": []
          }
        },
        "serve": {
          "builder": "@angular/build:dev-server",
          "options": {
            "buildTarget": "demo:build",
            "proxyConfig": "proxy.conf.json"
          }
        }
      }
    }
  }
}

Save tsconfig.json. Explicit .ts imports are rewritten during compilation:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "preserve",
    "moduleResolution": "bundler",
    "experimentalDecorators": true,
    "strict": true,
    "lib": ["ES2022", "DOM"],
    "types": [],
    "rewriteRelativeImportExtensions": true
  },
  "angularCompilerOptions": { "strictTemplates": true },
  "files": ["src/main.ts"]
}

Save src/index.html:

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <base href="/" />
    <title>Chunked Angular upload</title>
  </head>
  <body><app-root></app-root></body>
</html>

Save proxy.conf.json. The CLI’s Vite-based server needs /api/** to match nested upload routes; see Angular’s proxy configuration. The browser uses relative URLs, so requests remain on the application’s origin:

{
  "/api/**": { "target": "http://127.0.0.1:4317", "secure": false }
}

Save src/app/app.config.ts. Angular’s default Fetch backend does not emit upload progress; this application explicitly selects XHR. The demo header lets the receiver reject ordinary cross-site form submissions. It is public, not an authentication credential:

import { provideHttpClient, withInterceptors, withXhr } from '@angular/common/http'
import type { ApplicationConfig } from '@angular/core'

export const appConfig: ApplicationConfig = {
  providers: [
    provideHttpClient(
      withXhr(),
      withInterceptors([
        (request, next) => next(request.clone({ setHeaders: { 'X-Demo-Upload': '1' } })),
      ]),
    ),
  ],
}

Add the local receiver

Save receiver.mts. This is a loopback development receiver without login or user ownership. It accepts requests with the demo header and, when supplied, the configured browser origin. Any local process can send that header; do not expose this server or proxy it to a public service.

Request	Local receiver behavior
`PUT /api/uploads/{id}` with `{size}`	Create a session or return `{offset, complete}` for the same UUID and size.
`PUT /api/uploads/{id}/bytes`	Validate the range and body, write and flush the chunk, then acknowledge with 204.
`POST /api/uploads/{id}/complete`	Require every byte, publish `uploads/{id}.bin` without replacing an existing file, then return 204.
`DELETE /api/uploads/{id}`	Remove a partial file and retain a cancellation marker; return 409 if publication won the race.

State checks and disk mutations run synchronously in one process after each request body is read. A late chunk cannot recreate a canceled session. Each accepted file is between 1 byte and 100 MiB, each chunk is at most 1 MiB, and the demo retains at most 32 IDs, including canceled and completed ones, until restart. It saves arbitrary bytes without content inspection.

import {
  closeSync,
  fsyncSync,
  ftruncateSync,
  linkSync,
  mkdirSync,
  openSync,
  rmSync,
  statSync,
  writeSync,
} from 'node:fs'
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
import { resolve } from 'node:path'

type Session = { size: number; offset: number; complete: boolean; canceled: boolean }
class RequestError extends Error {
  readonly status: number
  constructor(status: number, message: string) {
    super(message)
    this.status = status
  }
}

const maxFile = 100 * 1024 * 1024
const maxChunk = 1024 * 1024
const sessions = new Map<string, Session>()
const directory = resolve('uploads')
const port = Number(process.env.PORT ?? '4317')
const appOrigin = process.env.APP_ORIGIN ?? 'http://127.0.0.1:4200'
if (!Number.isInteger(port) || port < 1 || port > 65535) {
  throw new Error('PORT must be between 1 and 65535.')
}
if (new URL(appOrigin).origin !== appOrigin || new URL(appOrigin).hostname !== '127.0.0.1') {
  throw new Error('APP_ORIGIN must be a loopback URL such as http://127.0.0.1:4200.')
}
mkdirSync(directory, { recursive: true, mode: 0o700 })

function reply(response: ServerResponse, status: number, value?: unknown): void {
  response.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' })
  response.end(value === undefined ? undefined : JSON.stringify(value))
}

async function body(request: IncomingMessage, limit: number): Promise<Buffer> {
  const parts: Buffer[] = []
  let length = 0
  // Keep the socket available for a 413 response when stopping the body reader early.
  for await (const part of request.iterator({ destroyOnReturn: false })) {
    if (!(part instanceof Buffer)) throw new RequestError(400, 'Expected binary request data.')
    length += part.length
    if (length > limit) {
      request.resume()
      throw new RequestError(413, 'Request body is too large.')
    }
    parts.push(part)
  }
  return Buffer.concat(parts)
}

function reserve(id: string, size: number): Session {
  const existing = sessions.get(id)
  if (existing) {
    if (existing.canceled) throw new RequestError(410, 'Upload was canceled.')
    if (existing.size !== size) throw new RequestError(409, 'Upload size conflicts.')
    return existing
  }
  if (sessions.size >= 32) throw new RequestError(429, 'Restart the demo receiver after 32 IDs.')
  const file = openSync(resolve(directory, `${id}.part`), 'wx', 0o600)
  closeSync(file)
  const session = { size, offset: 0, complete: false, canceled: false }
  sessions.set(id, session)
  return session
}

async function handle(request: IncomingMessage, response: ServerResponse): Promise<void> {
  if (
    request.headers['x-demo-upload'] !== '1' ||
    (request.headers.origin !== undefined && request.headers.origin !== appOrigin)
  ) {
    throw new RequestError(403, 'Use the local demo application.')
  }
  const match = /^\/api\/uploads\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})(\/bytes|\/complete)?$/u.exec(request.url ?? '')
  if (!match) throw new RequestError(404, 'Unknown upload route.')
  const id = match[1]
  if (!id) throw new RequestError(400, 'Missing upload ID.')
  const operation = match[2] ?? ''
  const part = resolve(directory, `${id}.part`)
  const output = resolve(directory, `${id}.bin`)

  if (request.method === 'PUT' && operation === '') {
    let input: unknown
    try {
      input = JSON.parse((await body(request, 512)).toString('utf8'))
    } catch (error) {
      if (error instanceof RequestError) throw error
      throw new RequestError(400, 'Expected JSON containing size.')
    }
    if (
      typeof input !== 'object' || input === null || !('size' in input) ||
      typeof input.size !== 'number' || !Number.isSafeInteger(input.size) ||
      input.size < 1 || input.size > maxFile
    ) {
      throw new RequestError(400, 'Choose a file between 1 byte and 100 MiB.')
    }
    const session = reserve(id, input.size)
    reply(response, 200, { offset: session.offset, complete: session.complete })
    return
  }

  // Read first, then recheck state: cancellation can arrive while a body is pending.
  const data = await body(request, operation === '/bytes' ? maxChunk : 512)
  let session = sessions.get(id)
  if (request.method === 'DELETE' && operation === '') {
    if (session?.complete) throw new RequestError(409, 'Upload was already published.')
    if (!session) {
      if (sessions.size >= 32) throw new RequestError(429, 'Restart the demo receiver after 32 IDs.')
      session = { size: 0, offset: 0, complete: false, canceled: true }
      sessions.set(id, session)
    }
    session.canceled = true
    rmSync(part, { force: true })
    reply(response, 204)
    return
  }
  if (!session || session.canceled) throw new RequestError(410, 'Upload is unavailable.')

  if (request.method === 'PUT' && operation === '/bytes') {
    const range = /^bytes (\d+)-(\d+)\/(\d+)$/u.exec(request.headers['content-range'] ?? '')
    if (!range) throw new RequestError(400, 'Expected Content-Range.')
    const start = Number(range[1])
    const end = Number(range[2]) + 1
    const total = Number(range[3])
    if (
      !Number.isSafeInteger(start) || !Number.isSafeInteger(end) ||
      !Number.isSafeInteger(total) || session.complete || start !== session.offset ||
      total !== session.size || end <= start || end > total || data.length !== end - start
    ) {
      throw new RequestError(409, 'Range or body does not match the committed offset.')
    }
    // These synchronous mutations cannot interleave in this single Node process.
    const file = openSync(part, 'r+')
    try {
      let written = 0
      while (written < data.length) {
        const count = writeSync(file, data, written, data.length - written, start + written)
        if (count === 0) throw new Error('Disk write made no progress.')
        written += count
      }
      ftruncateSync(file, end)
      fsyncSync(file)
      session.offset = end
    } finally {
      closeSync(file)
    }
    reply(response, 204)
    return
  }

  if (request.method === 'POST' && operation === '/complete') {
    if (!session.complete) {
      if (session.offset !== session.size || statSync(part).size !== session.size) {
        throw new RequestError(409, 'Upload still has missing bytes.')
      }
      // A hard link publishes on this filesystem without replacing an existing output.
      linkSync(part, output)
      session.complete = true
      rmSync(part, { force: true })
      console.log(`Published uploads/${id}.bin`)
    }
    reply(response, 204)
    return
  }
  throw new RequestError(405, 'Unsupported upload operation.')
}

const server = createServer((request, response) => {
  handle(request, response).catch((error: unknown) => {
    if (response.destroyed) return
    if (error instanceof RequestError) {
      reply(response, error.status, { message: error.message })
      return
    }
    console.error('Receiver could not process the upload.')
    reply(response, 500, { message: 'Receiver could not process the upload.' })
  })
})
server.requestTimeout = 30_000
server.on('error', () => {
  console.error('Receiver failed to listen. Check whether its port is already in use.')
  process.exitCode = 1
})
server.listen(port, '127.0.0.1', () => console.log(`Receiver ready on http://127.0.0.1:${port}`))

The receiver calls Node’s fsyncSync before advancing its in-memory offset. This supports retry after a lost response while the process is alive; it is not crash recovery. Completed .bin files remain on disk after restart, but their session metadata does not. Abandoned .part files also remain until you remove them with the receiver stopped. Keep this demo in its own folder.

Implementing file slicing

The service below slices one chunk at a time. end is exclusive for File.slice, but inclusive in the range header. A file of 2,097,155 bytes therefore ends with bytes 2097152-2097154/2097155, not a total inferred from the short final chunk.

Creating an RxJS pipeline for chunk uploads

Place the complete service in src/app/file-upload.service.ts. concatMap keeps one request in flight. Each resume establishes server state before sending more bytes; errors remain observable and leave the selected file available for another resume attempt.

import { HttpClient, HttpEventType } from '@angular/common/http'
import { Injectable, inject } from '@angular/core'
import {
  BehaviorSubject,
  EMPTY,
  Subscription,
  catchError,
  concatMap,
  concatWith,
  defer,
  finalize,
  range,
  tap,
} from 'rxjs'

interface UploadState {
  progress: number
  running: boolean
  canceling: boolean
  complete: boolean
  message: string
}

@Injectable({ providedIn: 'root' })
export class FileUploadService {
  readonly state = new BehaviorSubject<UploadState>({
    progress: 0,
    running: false,
    canceling: false,
    complete: false,
    message: 'Select a file.',
  })
  readonly #http = inject(HttpClient)
  #request = new Subscription()
  #file: File | null = null
  #id = ''
  #offset = 0
  readonly #chunkSize = 1024 * 1024

  select(file: File): void {
    if (this.state.value.canceling) return
    this.pause()
    this.#file = null
    if (file.size === 0 || file.size > 100 * 1024 * 1024) {
      this.state.next({
        progress: 0,
        running: false,
        canceling: false,
        complete: false,
        message: 'Choose a file between 1 byte and 100 MiB.',
      })
      return
    }
    this.#file = file
    this.#id = crypto.randomUUID()
    this.#offset = 0
    this.state.next({
      progress: 0,
      running: false,
      canceling: false,
      complete: false,
      message: 'Ready.',
    })
    this.resume()
  }

  resume(): void {
    const file = this.#file
    if (!file || this.state.value.running || this.state.value.canceling || this.state.value.complete) {
      return
    }
    const url = `/api/uploads/${this.#id}`
    this.state.next({ ...this.state.value, running: true, message: 'Uploading…' })
    this.#request = this.#http
      .put<unknown>(url, { size: file.size }, { timeout: 30_000 })
      .pipe(
        concatMap((status) => {
          if (
            typeof status !== 'object' ||
            status === null ||
            !('offset' in status) ||
            typeof status.offset !== 'number' ||
            !Number.isSafeInteger(status.offset) ||
            status.offset < 0 ||
            status.offset > file.size ||
            !('complete' in status) ||
            typeof status.complete !== 'boolean' ||
            (status.complete && status.offset !== file.size)
          ) {
            throw new Error('Invalid upload state')
          }
          this.#offset = status.offset
          this.progress(file)
          if (status.complete) {
            this.finished()
            return EMPTY
          }
          const offset = status.offset
          return range(0, Math.ceil((file.size - offset) / this.#chunkSize)).pipe(
            concatMap((index) =>
              defer(() => {
                const start = offset + index * this.#chunkSize
                const end = Math.min(start + this.#chunkSize, file.size)
                return this.#http
                  .put(`${url}/bytes`, file.slice(start, end), {
                    headers: { 'Content-Range': `bytes ${start}-${end - 1}/${file.size}` },
                    observe: 'events',
                    reportProgress: true,
                    timeout: 30_000,
                  })
                  .pipe(
                    tap((event) => {
                      if (event.type === HttpEventType.UploadProgress) {
                        this.progress(file, Math.min(event.loaded, end - start))
                      }
                      if (event.type === HttpEventType.Response) {
                        if (event.status !== 204) throw new Error('Chunk was not committed')
                        this.#offset = end
                        this.progress(file)
                      }
                    }),
                  )
              }),
            ),
            // Finalization runs once, including when the server already has every byte.
            concatWith(
              defer(() =>
                this.#http
                  .post(`${url}/complete`, {}, { observe: 'response', timeout: 30_000 })
                  .pipe(
                    tap((response) => {
                      if (response.status !== 204) throw new Error('Publication was not confirmed')
                      this.finished()
                    }),
                  ),
              ),
            ),
          )
        }),
        finalize(() => this.state.next({ ...this.state.value, running: false })),
      )
      .subscribe({
        error: () =>
          this.state.next({ ...this.state.value, message: 'Upload interrupted. Resume to retry.' }),
      })
  }

  pause(): void {
    if (this.state.value.canceling) return
    this.#request.unsubscribe()
    if (this.#file && !this.state.value.complete) {
      this.progress(this.#file)
      this.state.next({ ...this.state.value, running: false, message: 'Paused.' })
    }
  }

  cancel(): void {
    if (this.state.value.canceling) return
    this.pause()
    if (!this.#file || this.state.value.complete) return
    const url = `/api/uploads/${this.#id}`
    this.#file = null
    let confirmed = false
    this.state.next({
      progress: 0,
      running: true,
      canceling: true,
      complete: false,
      message: 'Canceling…',
    })
    this.#request = this.#http
      .delete(url, { observe: 'response', timeout: 30_000 })
      .pipe(
        tap((response) => {
          if (response.status !== 204) throw new Error('Cancellation was not confirmed')
          confirmed = true
        }),
        // Errors and interrupted requests both leave cleanup unconfirmed.
        catchError(() => EMPTY),
        // Publish one settled state so observers can safely select another file.
        finalize(() =>
          this.state.next({
            ...this.state.value,
            running: false,
            canceling: false,
            message: confirmed ? 'Canceled.' : 'Cancellation unconfirmed. Check your uploads.',
          }),
        ),
      )
      .subscribe()
  }

  destroy(): void {
    this.pause()
    // Destruction also interrupts cleanup; its finalizer reports the unconfirmed result.
    this.#request.unsubscribe()
  }

  private progress(file: File, loaded = 0): void {
    this.state.next({
      ...this.state.value,
      progress: Math.min(99, Math.floor((100 * (this.#offset + loaded)) / file.size)),
    })
  }

  private finished(): void {
    this.state.next({
      progress: 100,
      running: false,
      canceling: false,
      complete: true,
      message: 'Upload complete.',
    })
  }
}

Adding pause, resume, and cancellation

Unsubscribing aborts the active HTTP request. It does not undo bytes the server already accepted. Pause preserves the File and UUID in memory; resume reconciles the offset. Reloading loses this state, so this example does not resume across page reloads. Pause before choosing a replacement file. That starts a new UUID and leaves the old partial file on disk; cancel first if you want the receiver to remove it. Cancel attempts server cleanup; a network failure or a race with publication requires checking the receiver’s output directory and log. While cleanup is pending, pause, resume, cancel, and file selection are disabled and ignored by the service. Destroying the component interrupts cleanup and reports cancellation as unconfirmed.

Tracking progress

Replace src/app/app.ts with this standalone root component. The built-in control-flow syntax needs no NgIf; AsyncPipe is explicitly imported. Progress stays below 100 until publication is acknowledged. A resume can move progress backward when unacknowledged bytes need resending.

import { AsyncPipe } from '@angular/common'
import { Component, inject, OnDestroy } from '@angular/core'
import { FileUploadService } from './file-upload.service.ts'

@Component({
  selector: 'app-root',
  standalone: true,
  imports: [AsyncPipe],
  template: `
    @if (upload.state | async; as state) {
      <label>File <input type="file" (change)="select($event)" [disabled]="state.running" /></label>
      <progress aria-label="Upload progress" max="100" [value]="state.progress"></progress>
      <p role="status">{{ state.message }}</p>
      <button (click)="upload.pause()" [disabled]="!state.running || state.canceling">Pause</button>
      <button (click)="upload.resume()" [disabled]="state.running || state.canceling || state.complete">Resume</button>
      <button (click)="upload.cancel()" [disabled]="state.canceling || state.complete">Cancel</button>
    }
  `,
})
export class App implements OnDestroy {
  readonly upload = inject(FileUploadService)

  select(event: Event): void {
    if (!(event.target instanceof HTMLInputElement)) return
    const file = event.target.files?.[0]
    event.target.value = ''
    if (file) this.upload.select(file)
  }

  ngOnDestroy(): void {
    this.upload.destroy()
  }
}

Save src/main.ts to bootstrap the application:

import { bootstrapApplication } from '@angular/platform-browser'
import { App } from './app/app.ts'
import { appConfig } from './app/app.config.ts'

bootstrapApplication(App, appConfig).catch(() => console.error('Application startup failed.'))

Run and verify the upload

After saving every file, install and build from your parent directory. A failed install or build stops the chain, and the subshell returns you to your original directory:

(
  cd chunked-upload-app &&
  corepack yarn install &&
  corepack yarn build
)

Start the receiver in one terminal, from the same parent directory:

(
  cd chunked-upload-app &&
  node receiver.mts
)

In a second terminal, start the Angular development server:

(
  cd chunked-upload-app &&
  corepack yarn start
)

Open http://127.0.0.1:4200, select a nonempty file of at most 100 MiB, and watch the status. For a large transfer, click Pause and then Resume without reloading. A transfer can finish before you click; use browser network throttling if you want time to pause it. The progress bar stays below 100 until the receiver acknowledges publication. The page then shows Upload complete., and the receiver prints Published uploads/<UUID>.bin.

Compare that exact .bin file with the selected input using a checksum tool or a byte comparison. The UUID comes from the receiver’s log, not the original filename. Selecting the same input again starts a separate upload; the component clears the native input after selection so its change event can fire again. Stop both foreground servers with Ctrl+C.

Upload interrupted. Resume to retry. means a request failed or the receiver returned an invalid state. Check that the receiver is running and the proxy points to its port, then resume. If cancellation reports Cancellation unconfirmed. Check your uploads., inspect the receiver’s log and output directory: publication may have completed before cancellation arrived.

If the default ports are occupied, stop your own conflicting process or choose two free ports. Change the frontend port in package.json and the backend port in proxy.conf.json, then start the receiver with matching PORT and APP_ORIGIN values, such as PORT=4318 APP_ORIGIN=http://127.0.0.1:4201 node receiver.mts from the project directory. Restart the Angular server after changing its proxy configuration.

For a public application, the receiving API also needs authenticated ownership and validated CSRF protection on every mutation. The offset protocol here does not supply either. If you want a standard protocol and an established upload UI, explore tus and Uppy.