Rxjs Wait For Previous Execution To Finish

Ok, so I'm new to RxJs and I can't figure out some thing. I need to implement image processing in which the user adds multiple images at a time and for each image, among others, fo

Solution 1:

To achieve the following:

1. Thumbnails generation should be done sequentially order is important => use concatMap
2. Upload should be done as soon as the thumbnail is generated
3. Upload can be done in parallel order is not important => use map

So the final code could be

imageObservable
  .pipe(
    // preserve orders andwaitfor previous observable
    concatMap(generateThumbnail), 

    //map the generate
    map(upload)
  )
  .subscribe();

Note: upload does not wait for the generateThumbnail promise since this should be handled by concatMap itself.

asyncfunctionupload(image) {
  console.log(`[upload] start ${image.image}`);
  returnnewPromise(resolve => {
    setTimeout(() => {
      console.log(`[upload] done ${image.image}`);
      resolve();
    }, 1000);
  });
}

Solution 2:

Do you only want generating thumbnails to be sequential or does generation of a second thumbnail need to wait for upload of the first one to finish as well?

If you don't want to wait for upload, you just need to concat them:

import { of, from, } from'rxjs';
import { concatMap } from'rxjs/operators';

const imageObservable = of(1, 2, 3, 4, 5);

functiongenerateThumbnail(image) {
    console.log(`Generating thumbnail ${image}`);
    returnnewPromise(resolve => {
        setTimeout(() => {
            console.log(`Finished generating thumbnail ${image}`);
            resolve({
                image,
                thumbnail: `This is a thumbnail of image ${image}`
            });
        }, 1500);
    });
}

asyncfunctionupload(imagePromise) {
    const image = await imagePromise;
    console.log('Uploading', image);
    returnnewPromise(resolve => {
        setTimeout(() => {
            console.log('Finished uploading', image);
            resolve();
        }, 1500);
    });
}

// imageObservable.pipe(map(generateThumbnail)).subscribe(upload);

imageObservable.pipe(concatMap(image =>from(generateThumbnail(image)))).subscribe();

concatMap does what you want, and I just made an Observable out of your Promise with from creational operator, which I don't think is needed. But it's usually easier to work with Observables instead of Promises when you use RxJs.

functiongenerateThumbnail(image): Observable<any> {
    console.log(`Generating thumbnail ${image}`);
    returnfrom(
        newPromise(resolve => {
            setTimeout(() => {
                console.log(`Finished generating thumbnail ${image}`);
                resolve({
                    image,
                    thumbnail: `This is a thumbnail of image ${image}`
                });
            }, 1500);
        })
    );
}

Edit: Can you try this?

imageObservable
    .pipe(
        concatMap(image => generateThumbnail(image)),
        concatMap(image => upload(image))
    )
    .subscribe();