node.js - Nestjs 中的 Observables - 异步读取文件

Question

我正在尝试异步读取 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);
        });
    }
}

所以我想在 Controller 中订阅这个发出的可观察值，但是在我订阅并读取主题 (BehaviorSubject) 后，这些值才会被读取。我知道我在订阅和发送数据方面做了一些错误的事情，但无法理解我在哪里做错了。每次 Controller 打印'数据订阅未定义'，然后继续读取文件并发出可观察的

这是 Controller 数据

@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;

    }

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

Answer 1

问题是您实际上并没有在 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);
        }
    });
});