rxjs - observable 什么都不发出（没有完成？）

Question

我有一个 Angular 6 应用程序，我最初用一个 rest api 设置我的后端，但我开始转换部件以使用 socket.io。

当我从我的 rest api 返回数据时，以下工作：

this.http.get(api_url + '/versions/entity/' + entityId).pipe(
  mergeMap((versions:IVersion[]) => versions),
  groupBy((version:IVersion) => version.type),
  mergeMap(group => group.pipe(
    toArray(),
    map(versions=> {
      return {
        type: group.key,
        versions: versions
      }
    }),
    toArray()
  )),
  reduce((acc, v) => acc.concat(v), [])
);

特快路线：

router.get('/entity/:entityId', (req, res) => {
  const entityId = req.params.entityId;

  Version.getVersionsByEntity(entityId, (err, versions) => {
    if (err) {
      res.json({success: false, msg: err});
    } else {
      res.json(versions);
    }
  });
});

使用 mongoose 从我的 mongo 数据库中获取数据：

export function getVersionsByEntity(entityId, callback) {
  console.log('models/version - get versions by entity');

  Version.find({'entity.entityId': entityId})
          .exec(callback);
}

但是，当我使用 socket.io 进行完全相同的调用时，observable 不会返回任何内容。我猜是因为它永远不会完成？数据传输成功后，http 调用是否会发送“完成”消息？

从这个服务发送套接字：

getVersionsByEntity(entityId): Observable<IVersion[]> {
    // create observable to list to refreshJobs message
    let observable = new Observable(observer => {
      this._socketService.socket.on('versionsByEntity', (data) => {
        observer.next(data);
      });
    });

    this._socketService.event('versionsByEntity', entityId);

    return <Observable<IVersion[]>> observable;
  }

从服务器调用相同的猫鼬函数。

从 (socket) 服务返回的 observable 确实包含数据，只有在我添加 toArray() 函数后，它才会在我订阅时打印任何内容..

有人可以帮我解决这个问题并解释这背后的理论吗？不是完成了吗？ http 是否使用 Angular 发送“已完成”消息？

编辑：我创建了一个简单的 stackblitz，它正在做我想做的事，但我想从 _dataService 中删除 take(1)，因为我可能想要更新数据，所以我想保持可观察的打开 - @987654321 @

编辑 2：我用扫描运算符替换 toArray 很接近，但它似乎为数组发射了两次。 reduce() 发出正确的数据，但似乎只在完成时发出（如 toArray），所以它没有更好 - https://stackblitz.com/edit/rxjs-toarray-problem-tpeguu

Answer 1

处理事件流有点不同。我试图在下面提出一些代码：

getVersionsByEntity(entityId): Observable<IVersion[]> {
    return defer(() => {
        this._socketService.event('versionsByEntity', entityId);

        return fromEvent(this._socketService.socket, 'versionsByEntity')
            .pipe(take(1))
    }) as Observable<IVersion[]>;
}

主要思想是将所有内容包装在defer 中，这将使Observable 变得懒惰，并且仅在订阅时调用socketService.event（您的原始实现急切地立即调用socketService.event）。急切地实现可能会产生意想不到的后果——如果 Observable 订阅得太晚，很容易错过事件。

我还建议使用fromEvent Observable factory - 它处理事件侦听器的设置和拆除。

最后在第一次发射后完成 Observable 我添加了take(1) - 这会将发射次数限制为 1 并取消订阅 Observable。

Answer 2

除了@m1ch4ls 所说的之外，您还必须考虑toArray 的工作原理。 toArray 将 Observable 通知的所有数据转换为此类数据的数组。为了使其工作，Observable 必须完成。

Angular http 客户端返回的 Observable 总是在第一次通知之后完成，因此 toArray 有效。 socket.io 流在关闭时完成，因此在这样的流上使用 toArray 可能会产生永远不会从 Observable 中获取任何值的效果，以防流未关闭。

另一方面，如果您想在一个通知后关闭流，如果您使用take(1) 会发生这种情况，那么您最好考虑使用 http 请求。套接字流被认为是一种长期存在的通道，因此如果您必须始终在传输一条消息后关闭它们，则它不符合它们的性质。

评论后的更新版本

这是没有take的代码

 getData() {
    return this._dataService.getVersions().pipe(
      map(
        (versions: Array<any>) => versions.reduce(
          (acc, val) => {
            const versionGroup = (acc.find(el => el.type === val.type));
            if (versionGroup) {
              versionGroup.versions.push(val)
            } else {
              acc.push({type: val.type, versions: [val]})
            }
            return acc;
          }, []
        )
      ),
    )

要理解的关键点是，您的服务正在向您返回 Array 的东西。为了达到您的结果，您可以使用Array 方法直接处理Array，而无需使用Observable 运算符。请不要将我上面使用的分组逻辑的代码视为正确的实现 - 真正的实现可能会受益于使用 lodash 之类的东西进行分组，但我不想让事情过于复杂。

这是你的原始代码

getData() {
    return this._dataService.getVersions().pipe(
      mergeMap((versions: Array<any>) => versions),
      groupBy((version:IVersion) => version.type),
      mergeMap(group => group.pipe(
        toArray(),
        map(versions=> {
          return {
            type: group.key,
            versions: versions
          }
        }),
        toArray()
      )),
      reduce((acc, v) => acc.concat(v), [])
    )
  }

这是做什么的

1. 您创建了一个对象流，将第一个 mergeMap 应用于 服务返回的数组

2. 通过groupBy 运算符创建一个新的 Observable，它发出

3. 根据您的逻辑分组的对象，然后您再输入一秒钟 复杂的mergeMap 接受groupBy 发出的数组， 把它们变成一个对象流，只是为了立即转换它们 通过第一个 toArray 再次到一个数组，然后 转换为 {type: string, versions: []} 类型的 Object，到 然后最终再次调用toArray 来创建一个Array of Array

4. 您要做的最后一件事是运行reduce 来创建您的最终数组

为什么它只适用于take？因为groupBy 仅在其源 Observable 完成时才会执行，因此您想要对一组有限的事物进行分组是有道理的。 reduce Observable 运算符也是如此。

take 是完成源 Observable 的一种方式。