I am trying to write a code in angular 11 for a scenario like this -
I have list of files, and for every file I hit an api (say api1), i take an fileId from response and i pass it to another api (say api2),i want to keep on hitting the api2 every 3 seconds,unless i dont get the status="available" in the response. Once i get the available status, i no more need to hit the api2 for that fileId and we can start processing for the next file in loop.
This whole process for every file that I have.
I understand we can achieve this using rxjs operators like mergeMap or switchMap (as the sequence do not matter to me right now) . But i am very new to rxjs and not sure how to put it together.
This is what i am doing right now -
this.filesToUpload.forEach((fileItem) => {
if (!fileItem.uploaded) {
if (fileItem.file.size < this.maxSize) {
self.fileService.translateFile(fileItem.file).then( //hit api1
(response) => {
if (response && get(response, 'status') == 'processing') {
//do some processing here
this.getDocumentStatus(response.fileId);
}
},
(error) => {
//show error
}
);
}
}
});
getDocumentStatus(fileId:string){
this.docStatusSubscription = interval(3000) //hitting api2 for every 3 seconds
.pipe(takeWhile(() => !this.statusProcessing))
.subscribe(() => {
this.statusProcessing = false;
this.fileService.getDocumentStatus(fileId).then((response)=>{
if(response.results.status=="available"){
this.statusProcessing = true;
//action complete for this fileId
}
},(error)=>{
});
})
}
Here's how I might do this given the description of what you're after.
The thing that makes this work is that we only subscribe once (not once per file), and we let the operators handle subscribing and unsubscribing for everything else.
Then nothing happens until we subscribe. That way concat can do the heavy lifting for us. There's no need for tracking anything ourselves with variables like this.statusProessing or anything like that. That's all handled for us! It's less error prone that way.
// Create callList. This is an array of observables that each hit the APIs and only
// complete when status == "available".
const callList = this.filesToUpload
.filter(fileItem => !fileItem.uploaded && fileItem.file.size < this.maxSize)
.map(fileItem => this.createCall(fileItem));
// concatenate the array of observables by running each one after the previous one
// completes.
concat(...callList).subscribe({
complete: () => console.log("All files have completed"),
error: err => console.log("Aborted call list due to error,", err)
});
createCall(fileItem: FileItemType): Observable<never>{
// Use defer to turn a promise into an observable
return defer(() => this.fileService.translateFile(fileItem.file)).pipe(
// If processing, then wait untill available, otherwise just complete
switchMap(translateFileResponse => {
if (translateFileResponse && get(translateFileResponse, 'status') == 'processing') {
//do some processing here
return this.delayByDocumentStatus(translateFileResponse.fileId);
} else {
return EMPTY;
}
}),
// Catch and then rethrow error. Right now this doesn't do anything, but If
// you handle this error here, you won't abort the entire call list below on
// an error. Depends on the behaviour you're after.
catchError(error => {
// show error
return throwError(() => error);
})
);
}
delayByDocumentStatus(fileId:string): Observable<never>{
// Hit getDocumentStatus every 3 seconds, unless it takes more
// than 3 seconds for api to return response, then wait 6 or 9 (etc)
// seconds.
return interval(3000).pipe(
exhaustMap(_ => this.fileService.getDocumentStatus(fileId)),
takeWhile(res => res.results.status != "available"),
ignoreElements(),
tap({
complete: () => console.log("action complete for this fileId: ", fileId)
})
);
}