Получение ошибки при вызове метода внутри наблюдаемой, который возвращает другую наблюдаемую

Question

Я новичок в angular и наблюдаем. Я пробовал flatMap, swithMap и concatMap, но ничего не помогает. У меня ошибка « Вы указали« неопределенное »там, где ожидался поток. Вы можете предоставить Observable, Promise, Array или Iterable .» Когда код варианта 1 переключателя запускается методом обработки ввода, тогда как случай по умолчанию работает полностью нормально.

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);
        }
      );
  }
  }

обработка ввода .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.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;
      }
    }

Answer 2

inputProcessing функция в вашем сервисе должна возвращать наблюдаемую. В вашей реализации созданная наблюдаемая только излучает значение, в то время как concatMap ожидает наблюдаемой. Отсюда ошибка: вы указали «неопределенное» там, где ожидался поток. Вы можете предоставить 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 ... default).

Когда внутренние наблюдаемые дают значение, тогда returnObservable испускает это значение во внешнюю flatMap.