在 observable 中调用返回另一个 observable 的方法时出错

Question

我是角度和可观察的新手。我尝试过 flatMap、swithMap 和 concatMap，但没有任何帮助。我收到错误“您在预期流的位置提供了“未定义”。您可以提供 Observable、Promise、Array 或 Iterable。”当 switch case 1 的代码运行 inputprocessing 方法时，而 default case 完全正常。

app.component.ts

 export class AppComponent implements OnInit, AfterViewChecked, OnChanges {
  ProcessInput(input: string, session: string): void {
    this.appService
      .ProcessInput(input, session)//this works fine and returns the response.
      .pipe(
        concatMap((res: any) => {
          return this.inputProcessing.inputProcessing(res);
        })
      )
      .subscribe(
        res => {
          this.arr.push("TEST");
        },
        err => {
          console.log(err);
        }
      );
  }
  }

inputprocessing.service.ts

import "reflect-metadata";
import { Injectable } from "@angular/core";
import { Container, inject, injectable, AsyncContainerModule } from "inversify";
import { OutputResponse } from "../response";
import { PrescriptionStatus } from "src/Classes/PrescriptionStatus";
import { service } from "./service.service";
import { HttpClient } from "@angular/common/http";
import { CourierMode } from "../enums/CourierMode";
import { Observable, of } from "rxjs";
import { switchMap, flatMap, concatMap } from "rxjs/operators";
import { PrescriptionStatusMain } from "src/models/PrescriptionStatusResponse";

@Injectable({
  providedIn: "root"
})
export class InputprocessingService {
  constructor(private service: service,private http: HttpClient) {}

  responseText: string;
  returnMessage: string;
  inputProcessing(response: OutputResponse): Observable<any> {

    this.responseText = response.output.generic[0].text;
    //works completely fine with default case, but when case 1: runs it gave an error
    switch (this.responseText) {
      case "1": {

        const obsFail = new Observable(observer => {
          observer.next(this.returnMessage);
          observer.error();
          observer.complete();
        });

        this.service.getStatus().pipe(
          concatMap((res: StatusMain) => {
            return obsFail ;
          })
        );
        break;
      }

      default: {
        const obs = new Observable(observer => {
          observer.next(this.responseText);
          observer.complete();
        });
        return obs;
      }
    }
  }
}

service.ts

import * as request from "request";
import * as request_promise from "request-promise-native";
import { Injectable } from "@angular/core";
import { HttpClient } from "@angular/common/http";
import { StatusMain } from "src/models/StatusResponse";
import { Observable } from "rxjs";
@Injectable()
export class service {
  constructor(private http: HttpClient) {}

  getStatus(): Observable<StatusMain> {
    const body = { bdate: "1999/05/01" };
    return this.http.post<StatusMain>(
      "http://localhost:49995/api/Status",
      body
    );
  }
}

Answer 1

inputProcessing 服务中的函数需要返回一个 observable。通过您的实现，创建的 observable 只会发出值，而 concatMap 需要一个 observable。因此出现错误：您在预期流的位置提供了“未定义”。您可以提供 Observable、Promise、Array 或 Iterable。

试试这个：

 export class AppComponent implements OnInit, AfterViewChecked, OnChanges {
  ProcessInput(input: string, session: string): void {
    this.appService
      .ProcessInput(input, session)//this works fine and returns the response.
      .pipe(
        flatMap((res: any) => {
          return this.inputProcessing.inputProcessing(res);
        })
      )
      .subscribe(
        res => {
          this.arr.push("TEST");
        },
        err => {
          console.log(err);
        }
      );
   }
  }

为您服务：

@Injectable({
  providedIn: "root"
})
export class InputprocessingService {
  constructor(private service: service,private http: HttpClient) {}

  responseText: string;
  returnMessage: string;
  inputProcessing(response: OutputResponse): Observable<any> {

    this.responseText = response.output.generic[0].text;
    let returnObservable;

    //works completely fine with default case, but when case 1: runs it gave an error
    switch (this.responseText) {
      case "1": {

        const obsFail = new Observable(observer => {
          observer.next(this.returnMessage);
          observer.error();
          observer.complete();
        });

        returnObservable = this.service.getStatus().pipe(
          concatMap((res: StatusMain) => {
            return obsFail ;
          })
        );
        break;
      }

      default: {
        returnObservable = new Observable(observer => {
          observer.next(this.responseText);
          observer.complete();
        });
      }
    }

    return returnObservable;
  }
}

注意AppComponent 如何实现flatMap。 flatmap 必须返回一个 observable。因此inputProcessing 创建了一个局部变量 (returnObservable)，该变量包含对不同 observables 的引用（switch 语句：case1，case2...default）。

当内部 observable 发出值时，returnObservable 将该值发出到外部 flatMap。

Answer 2

inputprocessing.service.ts 这段代码解决了这个问题。

@Injectable({
      providedIn: "root"
    })
    export class InputprocessingService {
      constructor(private service: service,private http: HttpClient) {}

      responseText: string;
      returnMessage: string;
      inputProcessing(response: OutputResponse): Observable<any> {

        this.responseText = response.output.generic[0].text;
        let returnObservable;

        //works completely fine with default case, but when case 1: runs it gave an error
        switch (this.responseText) {
          case "1": {      
    //This line solved the error.
            return this.service.getStatus().pipe(map(res=>{return res}));
            break;
          }

          default: {
            returnObservable = new Observable(observer => {
              observer.next(this.responseText);
              observer.complete();
            });
          }
        }

        return returnObservable;
      }
    }