如何解决Angular Rxjs：让 concat 等待很长时间一些重构：

我正在尝试自动化更新程序后端中某些数据的过程。我正在使用我的 Angular 前端，在那里我创建了一个只有主用户可以访问的功能，这应该让他登录每个管理（租户），在那里它会下载一些里面有一些错误数据的对象，从谷歌服务询问正确数据并在后端更新数据，并为每个租户执行所有这些操作。

我想将这些操作中的每一个都编写为可观察对象并使用 concat 来按顺序执行所有操作，但是在我完成获取正确数据之前（我在水龙头中进行的操作），它已经尝试登录下一个租户，所以当它实际上拥有正确的数据时，他将无法将它们上传到后端，因为它会拒绝来自错误租户的数据。

我认为这个问题是由于水龙头中需要长时间的操作造成的（我需要做一些需要更多时间的事情），

这是我的代码片段（没有无关的东西）：

const obsList = [] as Observable<any>[];
this.assignedTenants.forEach(tenant => {
  const obsList2 = [] as Observable<any>[];
  obsList.push(this.authenticationService.login(new Credentials(usr,psw),tenant.id));
  obsList.push(this.structureService.getStructuresWithWrongAltitude()
    .pipe(tap(structuresReceived => {
      obsList2 = [] as Observable<any>[];
      if (structuresReceived != null && structuresReceived.length > 0) {
        structuresReceived.forEach(s => {
          this.getElevation(new google.maps.LatLng(s.centro.coordinates[0],s.centro.coordinates[1]))
            .then(a => {
              s.centroAltitudine = a;
              this.obsList2.push(this.structureService.putStructure(s));
            })
            .catch();
        });
      }
  })));
  obsList.push(forkJoin(obsList2)
    .pipe(tap(() => this.storageService.logout())));
});
concat(...obsList).subscribe();

如您所见，这段代码应该为每个租户创建并执行 3 个 observable，第一个用于登录，第二个用于获取错误数据，获取正确数据并准备第三个，这将更新数据。正如我所说，通常当从第二个可观察对象 getStructuresWithWrongAltitude 进入水龙头时，我已经可以通过使用它尝试登录其他租户的日志看到。

我的理论是，一旦它得到错误的数据，它就会尝试执行仍然无效的第三个 observable，然后转到下一个租户，但我不知道如何解决这个问题。

我需要一种方法让第二个 observable 在点击完成之前不发射，或者另一种方法来防止在其他操作完成之前继续进行连接

感谢您的帮助

编辑：

我能够通过将 getElevation（它返回一个承诺）设置为一个可观察列表来解决这个问题，这反过来又会创建一个新的可观察列表来保存数据。

正如我之前所说，我需要做一些非常相似的事情，不同的是这次水龙头实际上必须进行大量计算，这将花费很长时间，因此我将无法使用相同的修复程序，因此，我的问题仍然存在：我可以让 concat 等到水龙头完成吗？

编辑 2 进行澄清

正如我在上次编辑中所说的那样，通过将水龙头内的内容转换为其他可观察对象来解决该特定示例，但我在另一个函数中遇到了几乎相同的问题

此功能需要在文件夹中查找文件，然后才能上传

const folderInput = this.folderInput.nativeElement;
folderInput.onchange = () => {
  this.filesUploaded = folderInput.files;
  const obsList = [] as any[];

  this.assignedTenants.forEach(tenant => {
    const obsList2 = [] as Observable<any>[];

    obsList.push(this.authenticationService.login(new Credentials(usr,tenant.id));

    obsList.push(this.fileService.getAll()
      .pipe(
        tap(filesReceived => {
          if (filesReceived != null && filesReceived.length > 0) {
            console.log('upload picture: received list of files to update');

            let i = filesReceived?.length;
            filesReceived?.forEach(f => {
              const pathReceived = (f.originalFilename as string).substr(1).split('\\');

              let found = false;
              let index = -1;
              
              // searching the file in the folder
              //...
              
              if (found) {
                console.log('found a file');
                const selectedFile = this.filesUploaded[index];
                const formData = new FormData();
                formData.append('file',selectedFile,selectedFile.name);
                obsList2.push(this.fileService.updateFile(formData,f.id));
              }
              i--;
            });
            console.log('upload picture: updated obsList2');
            obsList.push(forkJoin(obsList2).subscribe(() => {
              console.log('upload picture: uploaded pictures');
              this.storageService.logout();
            }));
          }
      }))
    );
  });

  this.loadingIndicatorService.loading$.next(true);
  let counter = obsList.length;
  concat(...obsList).subscribe(() => {
    counter--;
    console.log('upload pictures: remaining phases: ' + counter);
    if (counter <= 0) {
      this.loadingIndicatorService.loading$.next(false);
  }
});
};
folderInput.click();

解决方法

如果我理解了问题的核心，我认为核心点是tap中的操作是异步的，所以如果你想等待的话，tap不是完全正确的操作符因为它的结果。所以你最好使用像 concatMap 这样的东西。我要做的另一件事是将 Promise 转换为 Observable，然后对它使用管道来执行 fetch 操作，调用 Google 服务，然后更新。最后一点是关于最后的 concat 的使用。这意味着您将依次攻击每个租户。如果这是您想要做的，那没问题。如果您认为可以并行进行，您可能需要考虑将 concat 替换为 forkJoin。

代码看起来像这样。

const obsList = [] as Observable<any>[];
this.assignedTenants.forEach(tenant => {
  obsList.push(
    // create here an Observable which executes login,fetch the wrong data,ask Google for the right data and update sequentially
    this.authenticationService.login(new Credentials(usr,psw),tenant.id)).pipe(
      concatMap(() => this.structureService.getStructuresWithWrongAltitude()),concatMap(structuresReceived => {
        const obsList3 = [] as Observable<any>[];
        if (structuresReceived != null && structuresReceived.length > 0) {
          structuresReceived.forEach(s => {
            // transform the Promise into an Observable using the from function
            // and then concatenate with the update operation
            obsList3.push(
              from(this.getElevation(new google.maps.LatLng(s.centro.coordinates[0],s.centro.coordinates[1]))).pipe(
                concatMap(a => {
                  s.centroAltitudine = a;
                  return this.structureService.putStructure(s)
                ),)
            )
          }
        }
        // execute the calls to Google in parallel and  (for each tenant)
        return forkJoin(obsList3)
      }),concatMap(() => this.storageService.logout())
    )
  });
});
concat(...obsList).subscribe();

简而言之：否

您可以永远永远让同步代码等待 javascript 中的异步代码。 JS 在一个线程上运行，如果您尝试这样做，您的程序将停止。 JS 确实有 async-await 使它看起来像同步代码在等待（但它只是将延续放在事件循环上，根本不等待）。

另一方面，您的 tap 中的所有同步代码都将在可观察管道的下一部分执行之前（100% 的时间）完成。

但是

好消息，你永远不需要

您永远不需要同步代码来等待 javascript 中的异步代码。如果您使用的是 observable，那么您就拥有了决定代码运行顺序所需的所有工具。

如果在您的 tap 中，如果您有 .then 或 .subscribe，则您可能做错了什么。在 RxJS 中，这有充分的理由被认为是代码异味。

您的代码（就目前而言）难以阅读，因此除了您尝试的内容的粗略介绍之外，很难获得更多信息。

我的理解是：

对于每个用户：

使用 id 登录用户
call this.fileService.getAll() // 这是作为登录用户完成的吗？您的服务是否会为您处理此问题？
在 0+ 个文件上调用 this.fileService.updateFile

这是一个粗略的做法。这肯定不会编译。此外，如果我对你的 observables 的功能有更多的了解，它可以被清理很多，但从上面显示的代码来看它们有点神秘。

from(this.assignedTenants).pipe(
  concatMap(tenant => concat(
    this.authenticationService.login(new Credentials(usr,tenant.id),this.fileService.getAll().pipe(
      switchMap(filesReceived => forkJoin(
        filesReceived.map(f => {
          //Code to get formData and such
          if(found){
            return this.fileService.updateFile(formData,f.id);
          }
          return null;
        }).filter(v => v != null)
      )
    ))
  )),).subscribe(result => {
  console.log("Result of forkjoin: ",result);
}

一些重构：

/*****
 * An Observable that gets all files,updates them,then completes
 *****/
function updateFiles(): Observable<any[]>{
  return this.fileService.getAll().pipe(
    // This map should turn every file received into either:
    //  1. A service call to update that file
    //  2. null
    map(filesReceived => filesReceived.map(f => {
      //Code to get formData and such
      if(found){
        return this.fileService.updateFile(formData,f.id);
      }
      return null;
    })),// Filter out null entries in our serviceCalls array
    map(serviceCalls => serviceCalls.filter(
      serviceCall => serviceCall != null
    )),// subscribe to all our service calls at once
    switchMap(serviceCalls => forkJoin(serviceCalls))
  );
}

from(this.assignedTenants).pipe(
  // ConcatMap won't start the second tenant until the first one's 
  // updateFiles() observable completes.
  concatMap(tenant => concat(
    this.authenticationService.login(new Credentials(usr,updateFiles()
  )),).subscribe({
  next: result => console.log("The Result of login(...) or updateFiles()",result),complete: () => console.log("Every file for every tenant is done")
})

Angular Rxjs：让 concat 等待很长时间 一些重构：

如何解决Angular Rxjs：让 concat 等待很长时间 一些重构：

解决方法

简而言之：否

但是

好消息，你永远不需要

一些重构：

相关推荐

Angular Rxjs：让 concat 等待很长时间一些重构：

如何解决Angular Rxjs：让 concat 等待很长时间一些重构：