如何将订阅结果传递至onNext？RxJS服务数据获取函数相关咨询

我实现了一个从Web服务获取信息的函数，代码如下：

const rxjs = require('rxjs');
const request = require('request');
const moment = require('moment');
function getAvailableGroupClasses() {
  return rxjs.Observable.create(observable => {
    request(`https://myservice.com/get.json`, (error, response, body) => {
      const j = JSON.parse(body);
      observable.next(j);
      observable.complete();
    });
  });
}

该函数返回一个Observable对象，后续会被另一个函数findClassOnDateTime...订阅。现咨询如何将订阅结果传递至onNext方法？

解决方案 #

其实你已经走在正确的路上啦！你在getAvailableGroupClasses里调用observable.next(j)的时候，就已经把解析后的JSON数据传递给订阅者的onNext（RxJS v5+里统一叫next）方法了。接下来只需要在订阅这个Observable时，通过正确的回调接收数据就行。

举个具体的实现例子，假设你的findClassOnDateTime函数是这样处理订阅的：

function findClassOnDateTime(targetDateTime) {
  // 订阅getAvailableGroupClasses返回的Observable
  getAvailableGroupClasses().subscribe({
    next: (classesData) => {
      // 这里的classesData就是你从服务拿到的j，也就是onNext接收的结果
      console.log('获取到的课程数据:', classesData);
      // 在这里可以根据targetDateTime做筛选逻辑
      const matchedClass = classesData.find(cls => 
        moment(cls.dateTime).isSame(targetDateTime, 'minute')
      );
      if (matchedClass) {
        console.log('找到匹配的课程:', matchedClass);
      } else {
        console.log('未找到符合时间的课程');
      }
    },
    error: (err) => {
      // 一定要处理错误！比如网络失败、JSON解析出错这些情况
      console.error('获取课程数据失败:', err);
    },
    complete: () => {
      console.log('课程数据获取流程完成');
    }
  });
}

几个重要的补充优化 #

• 完善错误处理：你当前的代码没有处理request的错误和JSON解析可能抛出的异常，这会导致Observable遇到问题时直接崩溃却没有提示，建议补上：

function getAvailableGroupClasses() {
  return rxjs.Observable.create(observable => {
    request(`https://myservice.com/get.json`, (error, response, body) => {
      if (error) {
        observable.error(error); // 把请求错误传递给订阅者的error回调
        return;
      }
      try {
        const j = JSON.parse(body);
        observable.next(j);
        observable.complete();
      } catch (parseErr) {
        observable.error(parseErr); // 捕获JSON解析失败的错误
      }
    });
  });
}

• RxJS版本适配：如果你的项目用的是RxJS 6及以上版本，更推荐用new Observable()的方式创建流，写法更符合新版本规范：

// RxJS 6+ 写法
const { Observable } = require('rxjs');

function getAvailableGroupClasses() {
  return new Observable(observable => {
    // 内部逻辑和上面带错误处理的版本一致
  });
}

这样一来，当findClassOnDateTime订阅Observable时，next回调（也就是你提到的onNext）就能稳稳接收到从Web服务获取并解析后的结果了。

内容的提问来源于stack exchange，提问作者Miguel Ribeiro