Dfyjkt

I was trying to get the fetched data based on the changed events. The result I was expected should be an object instead I see observable as the fetch produces promise.

The flatMap was used to convert promise to an object, but it seems I was doing wrong. Please let me know how I can fix it.

/** Receives UpdateEvent */
const changes$ = new Subject<FileChangesEvent>();

/** mock for 'this' */
const context = ;

/** Process FilesChange event */
const stream$ = changes$
 .pipe(
 /** Side effect for log current step */
 tap(event => console.log('onFileChanges trigged, event data: ', event.changes)),
 /** Fetching array of changes */
 map(event => event.changes),
 /** Use it for wait all 'forked' observables */
 flatMap(
 /** Process each record in parallel */
 changes => forkJoin(
 changes.map(change =>
 /** Update processing */
 of(change)
 .pipe(
 tap(input => console.log('Validating the file Type: ', input)),
 /** fetching data based on update type */
 flatMap(input =>
 input.type === FileChangeType.ADDED
 ? context.processAdds(input)
 : context.processUpdates(input),
 ),
 /** remove large files from update */
 map(input => (
 input.type === FileChangeType.UPDATED && input.size < CONTENT_SIZE_LIMIT
 ? Object.assign(input, value: '' )
 : input
 )),
 /** side effect for pubsub notifications */
 tap(updates => context.notify(updates)),
 ),
 ),
 ),
 ),
 );

/** When updates stream was finished */
stream$.subscribe(
 (...args) => console.log('[TICK]: ', ...args), // When all event updates was processed
 (...args) => console.log('[ERROR]: ', ...args),
 (...args) => console.log('[DONE]: ', ...args),
);

export const updateEventsStream = (event) => changes$.next(event);

I see two problems in your code. First is destructuring a promise - map(i => Object.assign(i, ...fetch(i.url))). It destroys it completely and you will not be able to get a result out of it. If you need to modify the value returned from api use then: map(i => fetch(i.url).then(val => Object.assign(i, ...val))).

Ok, next problem exactly with observables. You are correctly using flatMap to unwrap value returned from processAdds (and others), but the fact is these functions return promise wrapped in observable. You start with creating observable out of input: of(input) and then map input to promise. Hence when you unwrap result in flatMap you get promise not it's value.

Simple solution would be to unwrap promise inside helper functions using any <something>Map operators, e.g. the same flatMap:

processAdds: (input) =>
 of(input).pipe(delay(input.debounce || 1), flatMap(i => fetch(i.url))),

You can check full example here - https://stackblitz.com/edit/rxjs-fetch-flatmap?devtoolsheight=60