nestjs中的Observables

我正在尝试异步读取json文件并将其作为响应发送出去（作为rxjs可观察数据）的用例。这是我使用的服务

 import { logger } from './../../shared/utils/logger';
import { Injectable } from '@nestjs/common';
import * as fs from 'fs';
import * as path from 'path';
import { BehaviorSubject, Observable, pipe, of, from, throwError, merge} from 'rxjs';
import { map, filter, scan, take, debounce, switchMap, retry, catchError, mergeMap, delay, zip, tap, mapTo } from 'rxjs/operators';
import { HttpResponseModel } from '../model/config.model';
import { isNullOrUndefined } from 'util';
@Injectable()
export class NewProviderService {
    serviceSubject: BehaviorSubject<HttpResponseModel[]>;
    filePath: string;
    httpResponseObjectArray: HttpResponseModel[];
    constructor() {
        this.serviceSubject = new BehaviorSubject<HttpResponseModel[]>([]);
        this.filePath = path.resolve(__dirname, './../../shared/assets/httpTest.json');
        this.setSubject();
    }


 readFileFromJSON() {
      this.readFileFromJsonSync();
      fs.exists(this.filePath.toString(), exists => {
        if (exists) {
           fs.readFile(this.filePath.toString(), 'utf-8', (err, data) => {
                logger.info('file read without parsin', data);
                this.httpResponseObjectArray = JSON.parse(data).HttpTestResponse;
                logger.info('array obj is:', this.httpResponseObjectArray);
                logger.info('file read after parsing', JSON.parse(data));
                return this.httpResponseObjectArray;
            });
        } else {
            return null;
        }

    });
}


getObservable(): Observable<HttpResponseModel[]> {
       // create an observable
        // return Observable.create(observer => {
        //     observer.next(this.readFileFromJSON());
        // });

        return of(this.readFileFromJsonSync()).pipe(map(data => {
            logger.info('inside obs methid', data);
            return data;
        }));

    }

    setSubject() {
        this.getObservable().subscribe(data => {
            logger.info('data before setting in sub', data);
            this.serviceSubject.next(data);
        });
    }
}

所以我想在控制器中订阅这个发出的observable，但是在订阅和阅读主题（BehaviorSubject）之后，这些值会被读取。我知道我在订阅和发送数据方面做错了，但无法理解我做错了什么。每次控制器打印'订阅未定义的数据'，然后继续读取文件并发出可观察量

这是控制器数据

@Get('/getJsonData')
  public async getJsonData(@Req() requestAnimationFrame, @Res() res) {
    this.newService.serviceSubject.subscribe(data => {
      logger.info('data subscribed', data);
      res.status(HttpStatus.OK).send(data);
    });

  }

如果我同步读取文件，它会很好用

用以下方法替换readFileFromJSON()并且效果很好

readFileFromJsonSync(): HttpResponseModel[] {
        const objRead = JSON.parse(fs.readFileSync(this.filePath.toString(), {encoding: 'utf-8'}));
        logger.info('object read is', objRead.HttpTestResponse);
        return objRead.HttpTestResponse;

    }

所以我在读取文件异步时遗漏了一些东西。我不确定我做错了什么。有人可以帮忙吗？

问题是你实际上没有在readFileFromJSON中返回任何东西。它将异步运行fs.exists和fs.readFile以及相应的回调，但忽略回调的结果。

你应该使用Promises代替。您可以自己创建Promise，也可以使用像bluebird这样的库，将fs从基于回调的API转换为基于Promise的API。有关更多信息，请参阅this thread。

return new Promise(function(resolve, reject) {
    fs.readFile(this.filePath.toString(), 'utf-8', (err, data) => {
        if (err) {
            reject(err); 
        } else {
            const httpResponseObjectArray = JSON.parse(data).HttpTestResponse;
            resolve(httpResponseObjectArray);
        }
    });
});