feat: upload operator

README.md

@@ -126,6 +126,50 @@ export class AppComponent  {
 
 [**read more**](https://nils-mehlhorn.de/posts/angular-file-download-progress)
 
+### upload
+
+Transform HTTP events into an observable upload for indicating progress.
+
+`upload(): (source: Observable<HttpEvent<unknown>>) => Observable<Upload>`
+
+**Example**
+
+```typescript
+@Component()
+export class AppComponent {
+  upload$: Observable<Upload>;
+
+  constructor(private http: HttpClient) {}
+
+  upload(files: FileList | null) {
+    const file = files?.item(0);
+    if (!file) {
+      return;
+    }
+    const data = new FormData();
+    data.append("file", file);
+    this.upload$ = this.http
+      .post("/users/123/avatar", data, {
+        reportProgress: true,
+        observe: "events"
+      })
+      .pipe(upload());
+  }
+}
+```
+
+```html
+<input type="file" #fileInput (change)="upload(fileInput.files)" />
+<mat-progress-bar
+  *ngIf="upload$ | async as upload"
+  [mode]="upload.state == 'PENDING' ? 'buffer' : 'determinate'"
+  [value]="upload.progress"
+>
+</mat-progress-bar>
+```
+
+[**read more**](https://nils-mehlhorn.de/posts/angular-file-upload-progress)
+
 ### ignoreNotFound
 
 Ignores 404 error responses by instead completing the underlying observable.

projects/ngx-operators/src/lib/download.ts

@@ -1,25 +1,7 @@
-import {
-  HttpEvent,
-  HttpEventType,
-  HttpProgressEvent,
-  HttpResponse
-} from "@angular/common/http";
-import { Observable } from "rxjs";
-import { distinctUntilChanged, scan } from "rxjs/operators";
-
-function isHttpResponse<T>(event: HttpEvent<T>): event is HttpResponse<T> {
-  return event.type === HttpEventType.Response;
-}
-
-function isHttpProgressEvent(
-  event: HttpEvent<unknown>
-): event is HttpProgressEvent {
-  return (
-    event.type === HttpEventType.DownloadProgress ||
-    event.type === HttpEventType.UploadProgress
-  );
-}
-
+import { HttpEvent } from '@angular/common/http';
+import { Observable } from 'rxjs';
+import { distinctUntilChanged, scan } from 'rxjs/operators';
+import { isHttpProgressEvent, isHttpResponse } from './http';
 export interface Download {
   content: Blob | null;
   progress: number;

projects/ngx-operators/src/lib/http.ts

@@ -0,0 +1,16 @@
+import { HttpEvent, HttpResponse, HttpEventType, HttpProgressEvent } from "@angular/common/http";
+
+export function isHttpResponse<T>(
+  event: HttpEvent<T>
+): event is HttpResponse<T> {
+  return event.type === HttpEventType.Response;
+}
+
+export function isHttpProgressEvent(
+  event: HttpEvent<unknown>
+): event is HttpProgressEvent {
+  return (
+    event.type === HttpEventType.DownloadProgress ||
+    event.type === HttpEventType.UploadProgress
+  );
+}

projects/ngx-operators/src/lib/upload.spec.ts

@@ -0,0 +1,52 @@
+import { HttpEvent, HttpEventType, HttpResponse } from '@angular/common/http'
+import { Subject } from 'rxjs'
+import { toArray } from 'rxjs/operators'
+import { upload } from './upload'
+
+describe('upload', () => {
+  it('should transform HTTP upload', done => {
+    const request = new Subject<HttpEvent<unknown>>()
+    request.pipe(upload(), toArray())
+      .subscribe(uploads => {
+        expect(uploads).toEqual([
+          {progress: 0, state: 'PENDING'},
+          {progress: 1, state: 'IN_PROGRESS'},
+          {progress: 50, state: 'IN_PROGRESS'},
+          {progress: 100, state: 'IN_PROGRESS'},
+          {progress: 100, state: 'DONE'}
+        ])
+        done()
+      })
+    request.next({type: HttpEventType.Sent})
+    request.next({
+      type: HttpEventType.User,
+    })
+    request.next({
+      type: HttpEventType.UploadProgress,
+      loaded: 12,
+      total: 1024
+    })
+    request.next({
+      type: HttpEventType.UploadProgress,
+      loaded: 512,
+      total: 1024
+    })
+    request.next({
+      type: HttpEventType.User,
+    })
+    request.next({
+      type: HttpEventType.UploadProgress,
+      loaded: 1024,
+      total: 1024
+    })
+    request.next({
+      type: HttpEventType.UploadProgress,
+      loaded: 1024
+    })
+    request.next(new HttpResponse({
+      status: 200,
+      url: '/uploads/file.pdf'
+    }))
+    request.complete()
+  })
+})

projects/ngx-operators/src/lib/upload.ts

@@ -0,0 +1,39 @@
+import { HttpEvent } from "@angular/common/http";
+import { Observable } from "rxjs";
+import { distinctUntilChanged, scan } from "rxjs/operators";
+import { isHttpProgressEvent, isHttpResponse } from "./http";
+
+export interface Upload {
+  progress: number;
+  state: "PENDING" | "IN_PROGRESS" | "DONE";
+}
+
+export function upload(): (
+  source: Observable<HttpEvent<unknown>>
+) => Observable<Upload> {
+  const initialState: Upload = { state: "PENDING", progress: 0 };
+  const reduceState = (state: Upload, event: HttpEvent<unknown>): Upload => {
+    if (isHttpProgressEvent(event)) {
+      return {
+        progress: event.total
+          ? Math.round((100 * event.loaded) / event.total)
+          : state.progress,
+        state: "IN_PROGRESS",
+      };
+    }
+    if (isHttpResponse(event)) {
+      return {
+        progress: 100,
+        state: "DONE",
+      };
+    }
+    return state;
+  };
+  return (source) =>
+    source.pipe(
+      scan(reduceState, initialState),
+      distinctUntilChanged(
+        (a, b) => a.state === b.state && a.progress === b.progress
+      )
+    );
+}

projects/ngx-operators/src/public-api.ts

@@ -8,3 +8,4 @@ export * from "./lib/throw-for-codes";
 export * from "./lib/download";
 export * from "./lib/ignore-not-found";
 export * from "./lib/run-zone";
+export * from "./lib/upload";