bufferCount + concatMap: File chunking

user_5700239 0
Viewed 163

I am trying to build a component which basically does two things:

  1. Split the file into smaller blobs
  2. Upload the file parts, once all the parts are uploaded then make an API call and mark the item as upload completed.

So far, I have been able to create an end-to-end poc, but I am trying to improve on my code to upload only n chunks at a time then proceed to next batch and wait until all chunks are uploaded.

For the splitting logic I am using, bufferCount + forkJoin but I want to be able to call an API after all chunks are complete. Instead, it gets triggered after each batch completes.

  • The next batch should not get triggered if the previous batch fails.
const apiCallouts = parts.map((part, idx) => {
      const formData = new FormData();
      formData.append('part', part);
      const opts = { ...baseOptions(this.conf) };
      delete opts.headers['content-type'];
      return this.http.put(// Perform API Callout)
    });

const mergeObs = from(apiCallouts).pipe(
      bufferCount(5),
      concatMap(buffer => forkJoin(buffer))
    );

What would be the best way to go about this?

rxjs angular
1 Answers

Instead, it gets triggered after each batch completes.

I think for this you could use the toArray() operator:

const mergeObs = from(apiCallouts).pipe(
  bufferCount(5),
  concatMap(buffer => forkJoin(buffer)),

  // `forkJoin` in this case will return an array.
  // So, we can use `mergeAll()` to explode the array.
  mergeAll(),

  // Accumulate everything and emit on `complete`.
  toArray(),
);

The next batch should not get triggered if the previous batch fails.

I'd say here it depends on what you want to happen if a batch fails.

For example, if you want to end the entire stream, you could use something as follows:

/* ... */
concatMap(buffer => forkJoin(buffer)),
mergeAll(),
toArray(),

// `NEVER` will immediately emit a `complete` notification.
// If you want your `next` callback to be called, you can replace `NEVER`
// with something like `of('a message')`.
// Note: `err` would come from `concatMap`'s inner observable.
catchError(err => NEVER)

Or you might want to retry only that specific batch that failed:

/*  */

bufferCount(5),
concatMap(
  buffer => forkJoin(buffer).pipe(
    // Or, you could use `retryWhen` for more flexibility.
    retry(3),
  )
),

/* ... */
user_96326210
Related