Как сохранить поток после forkJoin или concatAll (), toArray ()

Question

Ради обучения я пытаюсь сохранить поток после, например, форк-джоина. Но я хотел бы сделать это с трубкой (элегантно). Не просто восстановить массив, а затем переписать из (массив) .subscribe, снова.

Я пытался сделать изнутри внутри трубы, но это не сработало. Также я хотел бы знать (в качестве бонусных вопросов), что было бы идеальным способом в трубе (в первом случае в данном случае, потому что я полагаю, что будет 2 трубы), чтобы изменить каждый результат вызова mock_post ДО массива forkJoin возвращается

Просто небольшая фиктивная функция, созданная мной для имитации возвращаемого значения API через 2 с.

    private mock_post(url: string): Observable<object> {
        return from(timer(2000)).pipe(
            map(result => {
                return {id: 2, name: 'mock', source: url};
            })
        );
    }

Первый тестовый пример с ForkJoin (с использованием вышеуказанной функции)

        const operations = [];
        operations.push(this.mock_post('/api/test/'));
        operations.push(this.mock_post('/api/test/2'));
        operations.push(this.mock_post('/api/test/3'));

        forkJoin(operations).pipe(

        )
            .subscribe(
                (next) => {
                    console.log(next);
                }
            );

Второй контрольный пример с concatAll () + toArray ().

    const obs: Observable<any> = from(operations);
        obs.pipe(
            map(result => {
                // I want to modify the result here, and add something to the object or modify a property (for example)
                console.log(`result-> `, result); // This is(obviously, returning an observable, but how do I get the final value here?)

                result.name = 'mock_modified';

                return result;
            }),
            concatAll(),
            toArray()
        ).subscribe(
            (next) => {
                console.log(`-> NEXT`, next);
            },
            (error) => {
                console.log(`* ERROR`, error);
            },
            () => {
                console.log('=) COMPLETE');
            }
        );

Если бы кто-нибудь мог пролить немного света здесь, было бы здорово. Надеюсь, я все правильно объяснил.

Answer 1

действительно нет необходимости выполнять операции с массивами с операторами.Вы можете просто сделать это на карте

 forkJoin(operations).pipe(            
            map((results: any) => 
               results.map(result=>({...result,name:'changed}))
),

Answer 2

Вот как может выглядеть ваш код:

function mock_post(url: string): Observable<any> {
  return timer(2000).pipe( // <-- you don't need to wrap 'timer' with 'from'
    map(result => ({ id: 2, name: 'mock', source: url }))
  );
}

const operations = [];
operations.push(mock_post('/api/test/'));
operations.push(mock_post('/api/test/2'));
operations.push(mock_post('/api/test/3'));

forkJoin(operations).pipe(
  map((responses: any[]) => responses.map(r => { // <-- Use Array.map to tranform the values
    console.log('result->', r);
    r.name = 'mock_modified';
    return r;
  }))
).subscribe(next => console.log('forkJoin', next));

concat(...operations).pipe( // <-- Use concat to create your Observable instead of piping to concatAll
  map((result: any) => { // <-- you can modify the results here
    console.log('result->', result);
    result.name = 'mock_modified';
    return result;
  }),
  toArray()
).subscribe(
  next => console.log('-> NEXT', next),
  error => console.log('* ERROR', error),
  () => console.log('=) COMPLETE')
);

Answer 3

Я отвечаю на свой вопрос, после поиска больше.

(кстати, я использую RXJS6.5)

для test1, после forkjoin, я в основном передал concatMap и внутри сделал from (результат), чтобы передать каждое значение из массива. После этого, сопоставляя каждый отдельный результат, просто протестируйте изменение этого возвращенного объекта и зарегистрируйте результат (вероятно, нажатие tap () достигло бы того же самого, если бы не было необходимости изменять объект.

После этого простой toArray () просто вернет мне 1 отдельный массив.

    test1() {

        const operations = [];
        operations.push(this.mock_post('/api/test/'));
        operations.push(this.mock_post('/api/test/2'));
        operations.push(this.mock_post('/api/test/3'));

        forkJoin(operations).pipe(
            concatMap((result) => {
                return from(result);
            }),
            map((result: any) => {
                console.log(`resulting-> `, result);
                result.name = 'changed';
                return result;
            }),
            toArray()
        ).subscribe(
            (next) => {
                console.log(next);
            }
        );
    }

Это сказал. Если у кого-то есть лучший способ или альтернативный способ.

Я весь в ушах.

Я отмечаю этот вопрос как отвеченный собой, потому что он работает (пока) так, как я хочу.

Я все еще понимаю все концепции RXJS и стараюсь изящно поддерживать поток.

Приветствие.