RXJS switchmap + коснитесь как оператор - PullRequest
Новая
       12

RXJS switchmap + коснитесь как оператор

1 голос
/ 10 июля 2019

У меня есть поток файлов, и я хочу заполнить дополнительную информацию о нем, но я хотел бы представить полученные в настоящее время данные пользователю, так как это все, что изначально видно в любом случае.

Я хочу заметить, что:

  • Отмена при новой эмиссии (например, switchMap)
  • Не ждет, пока наблюдаемое закончится, прежде чем испускать (например, tap)

То, что у меня сейчас есть, ждет результата, прежде чем отправлять файлы.

Настройка и текущая попытка изменения: 

this.pagedFLFiles = fileService.getFiles().pipe(
    switchMap(response => concat(
        of(response),
        fileService.getAdditionalInfo(response.items).pipe(
            switchMap(() => EMPTY),
        ),
    )),
    shareReplay(1),
);

fileService.getAdditionalInfo(response.items) - это изменение данных 

getAdditionalInfo(files: FLFile[]): Observable<FLFile[]> {
    return this.api.getWithToken(token => {
        return { path: `v5/user/${token}/files/${files.map(file => file.id).join(',')}}/facilities` };
    }).pipe(
        map(information => {
            files.forEach(file => {
                const info = information[file.id];
                (Object.entries(info) as [keyof typeof info, any][]).forEach(([key, value]) => {
                    file[key] = value;
                });
            });
            return files;
        }),
    );
}

1 Ответ

1 голос
/ 14 июля 2019

Используйте слияние вместо concat .

Concat ожидает как наблюдаемые, of (reponse), так и getAdditionalInfo, прежде чем выдавать значение.

Слияние генерируется каждый раз, когда генерируется одна из его наблюдаемых.

Пример: getFiles будет излучать каждую секунду в течение 3 секунд getAdditionalInfo будет отменен 2 раза (потому что он длится дольше 1 секунды), и поэтому будет изменять только массив последних выпущенных файлов 

import { merge, EMPTY, timer, of, interval } from 'rxjs';
import { finalize, switchMap, map, take, shareReplay } from 'rxjs/operators';


const fileService = {
  getFiles: () => interval(1000).pipe(
    take(3),
    map(x => {
      const items = [0, 1, 2].map(i => { return { 'info1': i }; })
      return { 'index': x, 'items': items };
    })
  ),
  getAdditionalInfo: (files) => {
    let wasModified = false;
    return timer(2000).pipe(
      map(information => {
        files.forEach(file => {
          file['info2'] = 'information' + files.length;
        });
        console.log('getAdditionalInfo: modified data');
        wasModified = true;
        return files;
      }),
      finalize(() => {
        if (!wasModified) {
          console.log('getAdditionalInfo: cancelled');
        }
      })
    );
  }
}


const pagedFLFiles = fileService.getFiles().pipe(
  switchMap(response => {
    return merge(
      of(response),
      fileService.getAdditionalInfo(response.items).pipe(
        switchMap(() => EMPTY),
      ));
  }
  ),
  shareReplay(1),
);


pagedFLFiles.subscribe(x => {
  console.log('immediate', x.index);
});

Stackblitz

...