angular - Δυναμική προσθήκη στην ουρά των RXJS Observables

Ο στόχος είναι να προστεθεί δυναμικά η ουρά των αιτήσεων HTTP και να χειριστεί η ουρά ως μία αίτηση. Ας δείξω πρώτα την τρέχουσα κατάσταση.

συστατικό.ts

public ngOnInit(): void {
    this.service.touched
        .pipe( 
             filter(res => !!res),
             takeUntil(this.ngUnsubscribe)
         )
        .subscribe(() => this.onSave());
}

public onSave(): void {
    this.busy = true;
    this.service.save()
        .pipe(finalize(() => this.busy=false))
        .subscribe(() => this.isInSync = true);
}

service.ts

public touched: BehaviorSubject<boolean> = new BehaviorSubject(false);
public saved: BehaviorSubject<boolean> = new BehaviorSubject(false);

public save(): Observable<ApiResponse> {

    const data = this.getDirtyData();

    const payload = data.map(( d: Data) => {
       /** [...] */
    });

    const reqOptions: HttpRequest<any> = this.api.setRequestOptions( '/url', payload );
    return this.api.apiHttpRequest(reqOptions)
        .pipe(
            map( resp => resp.data),
            tap( resp => this.saved.next(resp))
        );

}

Αυτό είναι που φαίνεται τώρα. Λειτουργεί καλά - αλλά αυτό που χρειάζομαι είναι να μπορώ να σχηματίσω μια ουρά από αυτά τα αιτήματα http αν ο χρήστης εκτελέσει τη μέθοδο onSave πολλές φορές, έτσι ώστε το component να κάνει finalize το Observable μόνο μία φορά. Αν το χρησιμοποιήσω όπως είναι, το component εγγράφεται στο observable πολλές φορές και έτσι θέτει τα this.busy και this.isInSync πρόωρα.

Οποιαδήποτε πρόταση για το πώς να το λύσουμε με κομψό τρόπο;

Απάντηση #1

Όταν κάνετε αυτό:

public onSave(): void {
    this.busy = true;
    this.service.save()
        .pipe(finalize(() => this.busy=false))
        .subscribe(() => this.isInSync = true); <--------------
}

Μπορείτε να το σκεφτείτε ως ένα σήμα προειδοποίησης στον κώδικα. Όταν καλείτε το subscribe, εκεί τελειώνει ο αντιδραστικός προγραμματισμός. Αυτό δεν θα έπρεπε να συμβεί μέχρι το πολύ τελευταίο σημείο αυτού που προσπαθείτε. Σε αυτή την περίπτωση, καθώς καλείτε το subscribe μέσα σε μια μέθοδο που καλείται πολλές φορές, θα χάσετε τη δυνατότητα να θέτετε τις αιτήσεις σε σειρά όπως θα θέλατε.

Νομίζω ότι αυτό που προσπαθείς να πετύχεις μπορεί να αναπαρασταθεί με τον ακόλουθο κώδικα (μείγμα mock για την κλήση HTTP και πραγματικής λογικής ροής):

const mockHttpRequest = (someParam: any) =>
  of(`Result: ${someParam}`).pipe(delay(1000));

const change$$ = new Subject<any>();

const queue$ = change$$.pipe(concatMap(change => mockHttpRequest(change)));

const isSyncing$ = combineLatest([
  change$$.pipe(
    map((_, index) => index),
    startWith(-1)
  ),
  queue$.pipe(
    map((_, index) => index),
    startWith(-1)
  )
]).pipe(
  map(([changeIndex, queueIndex]) => changeIndex !== queueIndex),
  distinctUntilChanged()
);

isSyncing$.subscribe(console.log);

Λίγη εξήγηση για αυτό:

const mockHttpRequest = (someParam: any) =>
  of(`Result: ${someParam}`).pipe(delay(1000));

Απλώς μιμηθείτε μια κλήση HTTP με καθυστέρηση

const change$$ = new Subject<any>();

Το άτομο που πρέπει να καλέσετε κάθε φορά που συμβαίνει μια αλλαγή

const queue$ = change$$.pipe(concatMap(change => mockHttpRequest(change)));

Όπως λέει το όνομά του: Αυτή είναι η ουρά για όλες τις κλήσεις HTTP. Θα εκτελούνται μία προς μία και δεν θα γίνονται παράλληλα.

const isSyncing$ = combineLatest([
  change$$.pipe(
    map((_, index) => index),
    startWith(-1)
  ),
  queue$.pipe(
    map((_, index) => index),
    startWith(-1)
  )
]).pipe(
  map(([changeIndex, queueIndex]) => changeIndex !== queueIndex),
  distinctUntilChanged()
);

Αυτό το παρατηρήσιμο θα σας ενημερώνει για την τρέχουσα κατάσταση της εφαρμογής: Εάν συγχρονίζει κάποια δεδομένα ή όχι. Για να το κάνουμε αυτό, συγκρίνουμε πόσα στοιχεία έχουν αποσταλεί επιτυχώς από την ουρά HTTP με το πόσες προσπάθειες αποθήκευσης έχουν γίνει. Εάν είναι ο ίδιος αριθμός, γνωρίζουμε ότι είμαστε συγχρονισμένοι, διαφορετικά συγχρονίζουμε.

Το distinctUntilChanged είναι απαραίτητο γιατί αν εκπέμψουμε, ας πούμε, 2 αλλαγές πολύ γρήγορα και προσπαθήσουμε να στείλουμε τη δεύτερη πριν επιστρέψει η πρώτη, το observable isSyncing$ θα εκπέμψει δύο φορές false πριν εκπέμψει true μόλις ολοκληρωθούν και οι δύο. Με το distinctUntilChanged διασφαλίζουμε ότι θα εκπέμψει μόνο μία φορά false και μόνο μία φορά true όταν ισχύει αυτή η κατάσταση.

Τέλος, για να δοκιμάσουμε αυτό, μπορούμε να κάνουμε τα εξής:

// simulate events happening when a user type at different times
// on the document to test our code
change$$.next('This is some initial text...');

setTimeout(() => {
  change$$.next('This is some initial text... And now with some other edit...');
}, 1500);

setTimeout(() => {
  change$$.next(
    'This is some initial text... And now with some other edit... And a quick one before previous is finished processing...'
  );
}, 1500);

setTimeout(() => {
  change$$.next(
    'This is some initial text... And now with some other edit... And a quick one before previous is finished processing... And a last one'
  );
}, 4500);

και θα πρέπει να εμφανίζει

false
true
false
true
false
true
false

Σημειώστε ότι αυτά που συμβαίνουν ταυτόχρονα (1500) δεν εκπέμπουν δύο φορές, απλώς η κατάσταση isSyncing$ παραμένει αληθής για λίγο περισσότερο από τα άλλα, καθώς οι δύο κλήσεις γίνονται η μία μετά την άλλη

false
true
false
true <--- this one
false
true
false

Αν θέλεις να παίξεις με αυτό, εδώ είναι ένα stackblitz: https://stackblitz.com/edit/rxjs-vtnk6z?devtoolsheight=60

Απάντηση #2

Ανάλογα με τον σχεδιασμό σας, combineLatest ή forkJoin μπορεί να είναι καλές επιλογές για εσάς. για παράδειγμα, αν έχετε περισσότερες από 1 κλήσεις υπηρεσιών και θέλετε να τις χειριστείτε ως μία κλήση αιτήματος:

forkJoin({
  serviceCall1: this.httpService1(input1:string...);
  serviceCall2: this.httpService2(input2:string...);
}).pipe().subscribe();

Το forkJoin περιμένει να ολοκληρωθούν όλα τα αιτήματα μέσα του και μετά εκπέμπει την τιμή τους. Το combineLatest εκπέμπει κάθε φορά που λύνεται ένα από τα αιτήματα.

Επίσης στο rxjs7 υπάρχουν connect, connectable που το λύνουν καλύτερα αλλά δεν είναι συμβατά με παλαιότερες εκδόσεις του Angular.

για περισσότερες πληροφορίες : forkJoin combineLatest