meteor-pubsub

用途

本技能用于编写和调试 Meteor 的发布（publication）与订阅（subscription），即 Meteor.publish 与 Meteor.subscribe。发布负责把一组文档流式推送给已订阅的客户端并保持实时更新；发布函数也是服务端唯一能在数据到达客户端之前进行行级授权过滤的位置。

决策流程

在决定使用发布还是方法时，可参考以下判断：

客户端是否需要该数据保持响应式？如果不需要，优先使用方法（一次性读取）；如果需要，则使用发布。
数据是否与用户相关？如果是，必须在 publish 函数内按 this.userId 过滤。缺少该过滤会导致文档在用户之间泄漏。
发布能否用单个游标表达？能则直接返回该游标。
是否需要在选择游标前做一次异步查询？使用异步发布处理器，await 查询结果后再返回游标。
是否需要逐文档的异步 join、自定义聚合输出或外部响应式数据源？此时应下沉到低层的 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 } },
  );
});

当集合包含订阅者不应收到的列时，应始终做字段投影。

异步发布处理器同样可以返回游标：

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 会先 await 该处理器，再处理其返回的游标。

订阅方式

import { Meteor } from "meteor/meteor";

const handle = Meteor.subscribe("items.mine");
// React / Blaze / Svelte 的 hooks 会观察 handle.ready() 与本地游标。
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

用于 join 或异步工作：

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 函数必须保持同步。逐文档的异步 join 应使用单独的发布或低层 API。

实时 observer 的投递不会等待前一个异步回调完成再处理后续变更。如果顺序或背压很重要，应使用按订阅的 Promise 队列串行化工作。同时应附加错误策略，并在 this.onStop 中停止 observer；详见 references/low-level-publish-api.md。

反模式

当数据与用户相关时，发布中缺少 this.userId 过滤。
发布敏感或不必要的列。当完整文档不属于发布契约时，应添加 fields 投影。
混淆异步发布处理器与异步游标 transform。Meteor 会 await 处理器的 Promise，但每个 transform 回调必须同步返回文档。
用游标 transform 做异步 join。transform 是同步的，低层 API 才是正确工具。
订阅大型无界集合。应使用 limit/skip 对数据分页。

参考

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