调用observable内部的方法时出错，该方法返回另一个observable

Question

我是新手，棱角分明，可观察。我已经尝试过flatMap，swithMap和concatMap，但没有任何帮助。我收到了错误消息“ 您在需要流的地方提供了'undefined'。您可以提供Observable，Promise，Array或Iterable。”当开关案例1的代码运行inputprocessing方法时，默认情况下完全可以正常工作。

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函数需要返回一个可观察值。在您的实现中，创建的可观察对象仅发出值，而concatMap期望可观察对象。因此出现错误：您在期望流的位置提供了“ undefined”。您可以提供一个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必须返回一个可观察值。因此，inputProcessing创建一个局部变量（returnObservable），该变量保存对不同可观察对象的引用（switch语句：case1，case2 ...默认）。

[当内部观察对象发出值时，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;
      }
    }