meteor-pubsub

Meteor のパブリケーションとサブスクリプション

パブリケーションはドキュメントのセットを購読中のクライアントにストリーミングし、最新の状態に保ちます。パブリケーションは、ドキュメントがクライアントに届く前にサーバー側の認可で行をフィルタリングできる唯一の場所です。

判断の流れ

1. クライアントはこのデータをリアクティブに必要としていますか？いいえの場合は、メソッド（1 回限りの読み取り）を優先します。はいの場合は、パブリケーションを使用します。
2. データはユーザー固有ですか？該当する場合は、publish 関数内で this.userId を使ってフィルタリングします。このフィルターがないと、ドキュメントがユーザー間で漏えいします。
3. データを単一のカーソルで表現できますか？可能な場合は、そのまま返します。
4. カーソルを選択する前に、非同期の検索が 1 回必要ですか？非同期の publish ハンドラーを使い、検索を await してからカーソルを返します。
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 で対象のフィールドを指定します。

非同期の publish ハンドラーからカーソルを返すこともできます。

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 を使って、コレクションごとに設定します。選択肢は 3 つあります。

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

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

戦略	使用する場面
SERVER_MERGE	デフォルトです。パブリケーション間で統合されたドキュメントのフィールドを追跡し、差分を送信します。
NO_MERGE	送信したドキュメント ID を追跡するため、購読解除時にそれらを削除できます。コレクションを 1 つのパブリケーションだけが所有する場合に使用します。
NO_MERGE_NO_HISTORY	何も記憶せず、停止時に削除も送信しません。クライアントに古いドキュメントが残っても問題ない、送信して終わるキューに限って使用します。

references/publication-strategies.md を参照してください。

低レベルの publish 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 の射影を追加します。
• 非同期の publish ハンドラーと非同期のカーソル transform を混同する。Meteor はハンドラーの Promise を await しますが、各 transform コールバックはドキュメントを同期的に返す必要があります。
• 非同期の結合にカーソル transform を使う。transform は同期処理です。低レベル API が適切なツールです。
• 上限のない大規模なコレクションを購読する。limit/skip でデータをページ分割します。

関連項目

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