Mongo.watch()

定義

Mongo.watch( pipeline, options )

レプリカセットとシャーディングされたクラスターのみ

レプリカセットまたはシャーディングされたクラスターの変更ストリーム カーソルを開き、 admin 、 local 、 configデータベースを除き、データベース全体のsystem以外のすべてのコレクションについてレポートします。 。

Parameter	タイプ	説明
pipeline	配列	任意。次の集計ステージの1つ以上で構成される集計パイプライン: $addFields $match $project $replaceRoot $replaceWith $redact $set $unset 変更イベント出力をフィルタリング/修正するためのパイプラインを指定します。 MongoDB 4.2 以降では、変更ストリームの集計パイプラインでイベントの _id フィールドが変更される場合、変更ストリームで例外がスローされるようになります。
options	ドキュメント	任意。 Mongo.watch()の動作を変更する追加オプション。

options ドキュメントには、次のフィールドと値を含めることができます。

フィールド	タイプ	説明
resumeAfter	ドキュメント	任意。 再開トークンで指定された操作の後に再開通知を試行するようにMongo.watch()に指示します。 各変更ストリームのイベント ドキュメントには、_id フィールドとして再開トークンが置かれます。変更後に再開する操作を表す変更イベント ドキュメントの _id フィールド全体を渡します。 resumeAfter は、startAfter および startAtOperationTime と排他関係にあります。
startAfter	ドキュメント	任意。 再開トークンで指定された操作の後に新しい変更ストリームの開始を試行するようにMongo.watch()に指示します。 無効化イベント 後に通知を再開できるようにします。 各変更ストリームのイベント ドキュメントには、_id フィールドとして再開トークンが置かれます。変更後に再開する操作を表す変更イベント ドキュメントの _id フィールド全体を渡します。 startAfter は、resumeAfter および startAtOperationTime と排他関係にあります。
fullDocument	string	任意。 デフォルトでは、アップデートされたドキュメント全体ではなく、アップデート操作によって変更されたフィールドのデルタが返されMongo.watch() 。 fullDocumentを"updateLookup"に設定すると、更新されたドキュメントの過半数がコミットした最新のバージョンを参照するようMongo.watch()に指示します。 Mongo.watch()は、 updateDescriptionデルタに加えて、ドキュメント検索を含むfullDocumentフィールドを返します。
batchSize	整数	任意。MongoDB クラスターからのレスポンスの各バッチで返す変更イベントの最大数を指定します。 cursor.batchSize() と同じ機能を持ちます。
maxAwaitTimeMS	整数	任意。空のバッチを返す前に、新しいデータ変更が変更ストリーム カーソルに報告されるまでサーバーが待機する最大時間（ミリ秒）。 デフォルトは 1000 ミリ秒です。
collation	ドキュメント	任意。 照合ドキュメントを渡し、変更ストリーム カーソルの照合順序を指定します。 省略した場合、デフォルトはsimpleバイナリ比較になります。
startAtOperationTime	タイムスタンプ	任意。変更ストリームの開始点。過去の開始点を指定する場合、oplog の時間範囲内である必要があります。oplog の時間範囲を確認するには、rs.printReplicationInfo() を参照してください。 startAtOperationTime は、resumeAfter および startAfter と排他関係にあります。

次の値を返します。	変更イベント ドキュメント上のカーソル。 変更イベント ドキュメントの例については、「変更イベント」を参照してください。

Tip

以下も参照してください。

db.collection.watch() および db.watch()

可用性

動作

• Mongo.watch() データを保持しているノードの大半に反映されたデータ変更についてのみ通知します。

• 変更ストリーム カーソルは、次のいずれかが発生するまで開いたままになります。

  • カーソルが明示的に閉じている。

  • 無効化イベントが発生している（コレクションの削除や名前の変更など）。

  • MongoDB 配置への接続が閉じているか、タイムアウトになっている。詳細については、「カーソルの動作」を参照してください。

  • 配置がシャーディングされたクラスターの場合、シャードを削除すると、開いている変更ストリームのカーソルが閉じることがあります。閉じた変更ストリームのカーソルは完全に再開できない場合があります。

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

npx mongodb-resumetoken-decoder <RESUME TOKEN>

アクセス制御

アクセス制御を使用して実行中の場合、ユーザーは すべてのデータベースにわたるすべての非システムfindchangeStream コレクション に対して および 特権アクションを持っている必要があります。つまり、ユーザーは、次の 特権 を付与する ロール を持っていなければなりません。

{ resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

組み込みの read ロールにより、適切な権限が付与されます。

カーソルの反復

MongoDB には、カーソルを反復処理する方法が複数用意されています。

cursor.hasNext() メソッドの場合、ブロックして次のイベントを待機します。watchCursor カーソルを監視してイベントを反復処理するには、次のように hasNext() を使用します。

while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}

cursor.tryNext() メソッドはノンブロッキングです。watchCursor カーソルを監視してイベントを反復処理するには、次のように tryNext() を使用します。

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

例

mongoshの次の操作は、レプリカセット上で変更ストリーム カーソルを開きます。 返されたカーソルは、system データベース、admin データベース、local データベースを除く全データベースにわたるすべてのconfig 以外のコレクションに対するデータ変更を報告します。

watchCursor = db.getMongo().watch()

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

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

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