Rxjs multi file upload queue with retry and stop behavior

Number 1:

1. If you click quickly more than 3 times on the "add file" button, only 3 files will appear in the queue. I'd like to have all of them but only 3 should be uploading at the same time. Yet, all the files to be uploaded should be displayed in the view, just waiting to start

First of all, this is a cool problem because as far as I could see, you can't simply compose the existing operators (Without getting stupid with partition). You need a custom operator that splits your stream. If you don't want to subscribe to your source twice, you should share before splitting.

There's quite a lot of work left to implement your solution the way you'd like. BUT, in terms of getting your stream to show all files regardless of whether they're currently loading, there's really just one piece missing.

You want to split your stream. One stream should emit default

{
  file,
  progress: 0,
  error: false
}`

files right away and the second stream should emit updates to those files. The second stream will have mergeAll(3), but the first doesn't need this limitation as it's not making a network request. You merge these two-streams and either update or add new entries into your output as you see fit.

Here's an example of that at work. I made a dummy example to abstract away the implementation details a bit. I start out with an array of objects with this shape,

{
  id: number,
  message: "HeyThere" + id,
  response: none
}

I make a fake httpRequest call that enriches an object to

{
  id: number,
  message: "HeyThere" + id,
  response: "Hello"
}

The stream emits each time a new object is added or when an object is enriched. But the enriching stream is limited to max 3 httpRequest calls at once.

const httpRequest= () => {
  return timer(4000).pipe(
    map(_ => "Hello")
  );
}

const arrayO = [];
arrayO.length = 10;
from(arrayO).pipe(
  map((val, index) => ({
    id: index,
    message: "HeyThere" + index,
    response: "None"
  })),
  share(),
  s => merge(s, s.pipe(
    map(ob => httpRequest().pipe(
      map(val => ({...ob, response: val}))
    )),
    mergeAll(3)
  )),
  scan((acc, val: any) => {
    acc.set(val.id, val);
    return acc;
  }, new Map<number, any>()),
  debounceTime(250),
  map(mapO => Array.from(mapO.values()))
).subscribe(console.log);

I added a debounce as I find it makes the output much easier to follow. Since I added all 10 un-enriched objects synchronously, it just spams 10 arrays to the output if I don't debounce. Also, since every fake HttpRequest takes exactly 4 seconds, I get three arrays spammed at the output every 4 seconds. Debounce stops the UI from stuttering or the console from getting spammed.

Number 2

The stop button should remove any upload. Whether it's uploading or failed

This is a can of worms because every canonical solution says you should make a state management system. That would be the easiest way to interact with files that are in Queue, Loading, Failed, and Loaded all in one uniform way.

It's pretty easy to implement a lightweight Redux-style state management system using RxJS (Just use scan to manage state and JSON objects representing events to transform state). The toughest part is managing your current httpRequests. You'd probably create a custom mergeAll() operator that takes in events, removes queued requests, and even cancels mid-flight requests if necessary.

Using a stopFile$$ works to cancel mid-flight requests but it'll fall apart if people want to stop a fileload that hasn't started yet (as per your first requirement, you want those vsible too). It's sort of brittle regardless because emiting on a suject never comes with the assurance that anybody is listening. Another reason that a redux-style management is the way to go.

This is a very interesting problem, here is my approach to it:

uploadFile$ = this.uploadFile.pipe(
    multicast(new Subject<CustomFile>(), subject =>
      merge(
        subject.pipe(
          mergeMap(
            // `file.id` might be created with uuid() or something like that
            (file, idx) =>
              of({ status: FILE_STATUS.PENDING, ...file }).pipe(
                observeOn(asyncScheduler),
                takeUntil(subject)
              )
          )
        ),

        subject.pipe(
          mergeMap(
            (file, idx) =>
              fakeUpload(file).pipe(
                map(progress => ({
                  ...file,
                  progress,
                  status: FILE_STATUS.LOADING
                })),
                startWith({
                  name: file.name,
                  status: FILE_STATUS.LOADING,
                  id: file.id,
                  progress: 0
                }),
                catchError(() => of({ ...file, status: FILE_STATUS.FAILED })),
                scan(
                  (acc, curr) => ({
                    ...acc,
                    ...curr
                  }),
                  {} as CustomFile
                ),
                takeUntil(
                  this.stopFile.pipe(
                    tap(console.warn),
                    filter(f => f.id === file.id)
                  )
                )
              ),
            3
          )
        )
      )
    )
  );
  files$: Observable<CustomFile[]> = merge(
    this.uploadFile$,
    this.stopFile
  ).pipe(
    tap(v =>
      v.status === FILE_STATUS.REMOVED ? console.warn(v) : console.log(v)
    ),
    scan((filesAcc, crtFile) => {
      // if the file is being removed, we need to remove it from the list
      if (crtFile.status === FILE_STATUS.REMOVED) {
        const { [crtFile.id]: _, ...rest } = filesAcc;

        return rest;
      }

      // simply return an updated copy of the object when the file has the status either
      // * `pending`(the buffer's length is > 3)
      // * `loading`(the file is being uploaded)
      // * `failed`(an error occurred during the file upload, but we keep it in the list)
      // * `retrying`(the `Retry` button has been pressed)
      return {
        ...filesAcc,
        [crtFile.id]: crtFile
      };
    }, Object.create(null)),
    // Might want to replace this by making the `scan`'s seed return an object that implements a custom iterator
    map(obj => Object.values(obj))
  );

StackBlitz demo.

I think the biggest problem here was how to determine when the mergeMap's buffer is full, so that a pending item should be shown to the user. As you can see, I've solved this using the multicast's second parameter:

multicast(new Subject(), subject => ...)

multicast(new Subject), refCount(), without its second argument, it's the same as share(). But when you provide the second argument(a.k.a the selector), you can achieve some sort of local multicasting:

  if (isFunction(selector)) {
    return operate((source, subscriber) => {
      // the first argument
      const subject = subjectFactory();
      
      /* .... */
      selector(subject).subscribe(subscriber).add(source.subscribe(subject));
    });
  }

selector(subject).subscribe(subscriber) will subscribe to the observable(which can also be a Subject) returned from the selector. Then, with .add(source.subscribe(subject)), the source is subscribed to. In the selector, we've used merge(subject.pipe(...), subject.pipe(...)), each of which will gain access to what's being pushed into the stream. Because of add(source.subscribe(subject)), the source's value will be passed to the Subject instance, which has its subscribers.

So, the way I solved the aforementioned problem was to create a race between observables. The first contender is

// #1
subject.pipe(
  mergeMap(
    // `file.id` might be created with uuid() or something like that
    (file, idx) =>
      of({ status: FILE_STATUS.PENDING, ...file }).pipe(
        observeOn(asyncScheduler),
        takeUntil(subject)
      )
  )
),

and the second one is

// #2
subject.pipe(
  mergeMap(
    (file, dx) => fileUpload().pipe(
      /* ... */
      // emits synchronously - as soon as the inner subscriber is created
      startWith(...)
    )
  )
)

So, as soon as the Subject(the subject variable in this case) receives the value from the source, it will send it to all of its subscribers - the 2 contenders. It all happens synchronously, which also means that the order matters. #1 will be the first subscriber to receive the value, and #2 will be second. The way the winner is selected is to see which one of the 2 subscribers emits first.

Notice that the first will pass along the value asynchronously(with the help of observeOn(asyncScheduler)) and the second one synchronously. The first one will emit first if the buffer is full, otherwise the second will emit.

I've ended up with the following:

export interface FileUpload {
  file: File;
  progress: number;
  error: boolean;
  toRemove: boolean;
}

export const uploadManager = () => {
  const file$$: Subject<File> = new Subject();
  const retryFile$$: Subject<File> = new Subject();
  const stopFile$$: Subject<File> = new Subject();

  const fileStartOrRetry$: Observable<File> = file$$.pipe(
    mergeMap(file =>
      retryFile$$.pipe(
        filter(retryFile => retryFile === file),
        startWith(file)
      )
    ),
    share()
  );

  const addFileToQueueAfterStartOrRetry$: Observable<
    FileUpload
  > = fileStartOrRetry$.pipe(
    map(file => ({
      file,
      progress: 0,
      error: false,
      toRemove: false
    }))
  );

  const markFileToBeRemovedAfterStop$: Observable<FileUpload> = stopFile$$.pipe(
    map(file => ({
      file,
      progress: 0,
      error: false,
      toRemove: true
    }))
  );

  const updateFileProgress$: Observable<FileUpload> = fileStartOrRetry$.pipe(
    map(file =>
      uploadMock().pipe(
        map(progress => ({ progress })),
        takeUntil(
          stopFile$$.pipe(filter(stopFile => stopFile.name === file.name))
        ),
        catchError(() => of({ error: true })),
        scan(
          (acc, curr: { progress: number } | { error: true }) => ({
            ...acc,
            ...curr
          }),
          {
            file,
            progress: 0,
            error: false,
            toRemove: false
          }
        )
      )
    ),
    // 3 upload in parallel maximum
    mergeAll(3)
  );

  const files$: Observable<FileUpload[]> = merge(
    addFileToQueueAfterStartOrRetry$,
    updateFileProgress$,
    markFileToBeRemovedAfterStop$
  ).pipe(
    scan<FileUpload, { [key: string]: FileUpload }>((acc, curr) => {
      if (curr.toRemove) {
        const copy = { ...acc };
        delete copy[curr.file.name];
        return copy;
      }

      return {
        ...acc,
        // todo we can't use the File reference directly here
        // but we shouldn't use the file name either
        // instead we should generate a unique ID for each upload
        [curr.file.name]: curr
      };
    }, {}),
    map(fileEntities => Object.values(fileEntities))
  );

  return {
    files$,
    file$$,
    retryFile$$,
    stopFile$$
  };
};

It covers all the cases as demonstrated here: https://rxjs-upload-multiple-file-v3.stackblitz.io

The code is here: https://stackblitz.com/edit/rxjs-upload-multiple-file-v3?file=src/app/upload-manager.ts

It's based on Mrk Sef's suggestion. It clicked after he mentioned "You want to split your stream".