db.collection.watch()

配置

db.collection.watch() は次のように、レプリカ セットおよびシャーディングされたクラスターの配置で利用できます。

• レプリカセットの場合、データを保持している任意のメンバーに対してdb.collection.watch()を発行できます。

• シャーディングされたクラスターの場合、 インスタンスでdb.collection.watch() mongosを発行する必要があります。

ストレージ エンジン

db.collection.watch()はWired Tiger ストレージ エンジンでのみ使用できます。

読み取り保証 (read concern)majority サポート

MongoDB 4.2 以降では、"majority" の読み取り保証 (read concern)のサポートに関係なく変更ストリームが利用可能になりました。つまり、変更ストリームを使用する際、読み取り保証 (read concern) のmajority サポートを有効にする（デフォルト）か無効にするかを選択できます。

MongoDB 4.0 以前では、変更ストリームは、"majority" 読み取り保証（read concern）サポートが有効（デフォルト）の場合にのみ使用できます。

再開可能性

MongoDBドライバーとは異なり、 mongoshはエラー後に変更ストリーム カーソルの再開を自動的に試行しません。 MongoDB ドライバーは、特定のエラーが発生後に 変更ストリーム カーソルの自動的な再開を1 回試行します。

db.collection.watch()は、oplog に保存されている情報を使用して変更イベントの説明を生成し、その操作に関連付けられた再開トークンを生成します。 resumeAfterまたはstartAfterオプションに渡される再開トークンによって識別される操作がすでにoplogから削除されている場合、 db.collection.watch()は変更ストリームを再開できません。

変更ストリームの再開の詳細については、「変更ストリームの再開」を参照してください。

アップデート 操作の完全なドキュメント検索

デフォルトでは、変更ストリーム カーソルはアップデート操作における特定のフィールドの変更またはデルタを返します。また、変更されたドキュメントのうち、過半数がコミットした最新のバージョンを検索して返すように変更ストリームを構成できます。アップデートと検索の間に他の書き込み操作が行われた場合、返されるドキュメントがアップデート実行時のドキュメントと大幅に異なる可能性があります。

アップデート操作中に適用された変更の数と完全なドキュメントのサイズによっては、アップデート操作における変更イベント ドキュメントのサイズが BSON ドキュメントの制限である 16 MB を超えるリスクがあります。サイズが超過した場合、サーバーで変更ストリーム カーソルが閉じられ、エラーが返されます。

変更ストリームを開く

次の操作は、 data.sensors コレクションに対し変更ストリーム カーソルを開きます。

watchCursor = db.getSiblingDB("data").sensors.watch()

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.tryNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

変更ストリーム出力の詳細なドキュメントについては、 「変更イベント」を参照してください。

変更ストリームで updateLookup に fullDocument オプションを使用する

fullDocument オプションを "updateLookup" に設定すると、アップデートされた変更ストリーム イベントに関連するドキュメントについて、過半数がコミットした最新のバージョンを参照するよう、変更ストリーム カーソルに指示します。

次の操作は、 fullDocument : "updateLookup"オプションを使用して、data.sensors コレクションに対し変更ストリーム カーソルを開きます。

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.tryNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

どのアップデート操作も、変更イベントはドキュメント検索の結果を fullDocument フィールドに返します。

fullDocumentのアップデート出力の例については、「変更ストリーム アップデート イベント」を参照してください。

変更ストリーム出力の詳細なドキュメントについては、 「変更イベント」を参照してください。

変更ストリームで集約パイプライン フィルターを使用する

次の操作は、集約パイプラインを使用して data.sensors コレクションに対する変更ストリーム カーソルを開き、insert イベントのみをフィルタリングします。

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.hasNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

変更ストリーム カーソルは、 operationType が insert である変更イベントのみを返します。変更ストリームの出力における詳細なドキュメントについては、「変更イベント」を参照してください。

変更ストリームの再開

変更ストリーム カーソルによって返されるすべてのドキュメントには、_id フィールドとして再開トークンが含まれます。変更ストリームを再開するには、再開する変更イベントの _id ドキュメント全体を watch() の resumeAfter または startAfter オプションに渡します。

次の操作では、再開トークンを使用して data.sensors コレクションに対する変更ストリーム カーソルを再開します。再開トークンを生成した操作が、クラスターの oplog からロール オフされていないことを前提としています。

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;
while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}
watchCursor.close();
let resumeToken = firstChange._id;
resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

カーソルを反復処理し、新しいイベントをチェックする。cursor.isClosed() メソッドを cursor.hasNext() メソッドと組み合わせて使用し、変更ストリーム カーソルが閉じられ、かつ最新のバッチにオブジェクトが残っていない場合にのみループが終了するようにします。

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

変更ストリームの再開に関する詳細なドキュメントについては、「変更ストリームの再開」を参照してください。