Ждите завершения нескольких наблюдателей

Question

Я работаю над проектом Angular7, который отображает информацию, поступающую с нескольких конечных точек старого, медленного бэкэнда.

• Я хочу обрабатывать и отображать информацию от каждой конечной точки по мере ее поступления.Таким образом, пользователю не нужно ждать самой медленной конечной точки, и нагрузка обработки остается распределенной.

• Когда самая медленная конечная точка наконец возвращается, тогда объединенный набор должен быть обработан коллективно.

• Порядок прибытия не гарантируется.

Воображаемый пример:

-У меня есть эти 2 источника данных Прямоугольники и квадраты.

-Расчет площади каждого предмета - тяжелая операция.

-Данные поступают в полных наборах, а не по элементам.

-Я хочу отсортировать полный набор при поступлении последнего набора.

    Rectangles: [ {x:1 , y:3} , {x:3,y:2}   ] 30 seconds to get this
    Squares   : [ { side:10 } , { side: 2} ]  2 minutes to get this

Desired output:
    Displayed at 30 seconds: Rect-Area-3 , Rect-Area-6

    Displayed at 2 minutes: Rect-Area-3 , Square-Area-4 , Rect-Area-6 , Square-Area-100

Текущий код

Прямо сейчас у меня есть такой код для каждого источника данных.Данные из каждого источника имеют совершенно разную структуру json, у них мало общего.

public getDemographicData(){
        const myObserver = {
            next: serverData => {
                //
                //Process result , save.
                //Let another component handle data presentation when 
                //change detection occurs
            },
            error: err => console.error('Observer got an error: ' + JSON.stringify(err)),
            complete: () => {}
        };

        // getDemographicsData returns an observable using http.get
        this.dataService.getDemographicsData(surveyId).subscribe(myObserver);
    }

Я не думаю, что здесь требуется forkjoin, так как можно сделать наблюдаемое из нескольких наблюдаемых.Я не хочу ждать поступления всех данных, прежде чем обрабатывать их.

Любая помощь или комментарии приветствуются, я не мог найти удовлетворительный или понятный ответ в другом месте.Я пишу только на сайтах помощи в качестве крайней меры.

Answer 1

Вам нужен forkJoin.

Выполните обработку данных отдельных потоков внутри tap каждого потока и запустите комбинированные вычисления в forkJoin, а затем подпишитесь на оба.

let fastObs = of('Fast').pipe(delay(1000), tap(console.log)); // One of the faster http requests, console.log will trigger after 1 second
let slowObs = of('Slow').pipe(delay(10000), tap(console.log)); // Slower, triggers after 10 seconds

forkJoin(fastObs, slowObs).subscribe(console.log); // Triggers after 10 seconds when you have both streams completed

https://stackblitz.com/edit/typescript-qrephm?file=index.ts