meteor-pubsub

Meteor 发布和订阅

发布会将一组文档流式传输给已订阅的客户端，并保持实时更新。发布是服务端授权在文档到达客户端前筛选行的唯一位置。

决策流程

1. 客户端是否需要响应式地获取这些数据？如果不需要，优先使用方法（一次性读取）。如果需要，则使用发布。
2. 数据是否因用户而异？在发布函数中使用 this.userId 进行筛选。若没有此筛选，文档就会泄露给其他用户。
3. 能否将发布表示为单个游标？直接返回该游标。
4. 是否需要先进行一次异步查询，再选择游标？使用异步发布处理程序，等待查询完成，然后返回游标。
5. 是否需要逐文档异步联接、自定义聚合输出或外部响应式数据源？改用底层的 this.added / this.changed / this.removed API。

脚手架

import { Meteor } from "meteor/meteor";
import { Items } from "/imports/api/items";

Meteor.publish("items.mine", function () {
  if (!this.userId) {
    return this.ready();
  }
  return Items.find(
    { ownerId: this.userId },
    { fields: { title: 1, qty: 1, updatedAt: 1 }, sort: { updatedAt: -1 } },
  );
});

如果集合包含订阅者不应接收的列，请通过 fields 指定投影。

异步发布处理程序也可以返回游标：

Meteor.publish("items.byTeam", async function (teamId) {
  const member = await Memberships.findOneAsync({
    teamId,
    userId: this.userId,
  });
  if (!member) return this.ready();
  return Items.find({ teamId }, { fields: { title: 1, qty: 1 } });
});

Meteor 会等待处理程序完成后，再处理返回的游标。

订阅

import { Meteor } from "meteor/meteor";

const handle = Meteor.subscribe("items.mine");
// React / Blaze / Svelte hooks observe handle.ready() and the local cursor.
handle.stop();

发布策略

使用 Meteor.server.setPublicationStrategy 按集合设置。共有三种选项：

import { DDPServer } from "meteor/ddp-server";

Meteor.server.setPublicationStrategy(
  "items",
  DDPServer.publicationStrategies.NO_MERGE,
);

策略	适用场景
SERVER_MERGE	默认策略。跟踪各发布之间合并后的文档字段，并发送增量更新。
NO_MERGE	跟踪已发送的文档 ID，以便在取消订阅时将其移除。适用于集合由单个发布管理的情况。
NO_MERGE_NO_HISTORY	不记住任何内容，停止时也不发送移除通知。仅用于发送后即忘的队列，且允许客户端保留过期文档。

参见 references/publication-strategies.md。

底层发布 API

适用于联接或异步处理：

Meteor.publish("feed", async function () {
  const cursor = Posts.find({}, { fields: { title: 1, authorId: 1 } });
  const observer = await cursor.observeChangesAsync({
    added: async (id, doc) => {
      const author = await Users.findOneAsync(doc.authorId, {
        fields: { username: 1 },
      });
      this.added("feed", id, { ...doc, authorName: author?.username });
    },
    changed: (id, changes) => this.changed("feed", id, changes),
    removed: (id) => this.removed("feed", id),
  });

  this.ready();
  this.onStop(() => observer.stop());
});

注意：游标的 transform 函数仍是同步的。逐文档异步联接应使用单独的发布或底层 API。

实时观察器的传递不会等待某个异步回调完成后再接收后续变更。如果顺序或背压很重要，请使用每个订阅独立的 Promise 队列串行处理任务。添加错误处理策略，并在 this.onStop 中停止观察器；参见 references/low-level-publish-api.md。

反模式

• 发布因用户而异的数据时，没有使用 this.userId 进行筛选。
• 发布敏感或不必要的列。如果完整文档不属于发布契约的一部分，请添加 fields 投影。
• 将异步发布处理程序与异步游标变换混为一谈。Meteor 会等待处理程序返回的 Promise，但每个变换回调都必须同步返回文档。
• 使用游标变换进行异步联接。变换是同步的；底层 API 才是合适的工具。
• 订阅大型且无界的集合。使用 limit / skip 对数据分页。

另请参阅

• references/publication-strategies.md
• references/low-level-publish-api.md
• references/eval-cases.md