Overview
このガイドでは、変更ストリームを使用してデータに対するリアルタイムの変更を監視する方法を学習できます。 変更ストリームは、アプリケーションがコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。
Rubyドライバーを使用する場合、watch メソッドを呼び出して Mongo::Collection::View::ChangeStreamオブジェクトを返すことができます。次に、そのコンテンツを反復処理して、アップデート、挿入、削除などのデータ変更を監視できます。
サンプル データ
このガイドの例では、Atlasサンプルデータセット の sample_restaurantsデータベースの restaurantsコレクションを使用します。Rubyアプリケーションからこのコレクションにアクセスするには、Atlas クラスターに接続する Mongo::Clientオブジェクトを作成し、次の値を database 変数と collection 変数に割り当てます。
database = client.use('sample_restaurants') collection = database[:restaurants]
MongoDB Atlasクラスターを無料で作成して、サンプルデータセットをロードする方法については、MongoDBを使い始めるガイドを参照してください。
変更ストリームを開く
変更ストリームを開くには、watch メソッドを呼び出します。watch メソッドを呼び出すオブジェクトによって、変更ストリームが監視するイベントの範囲が決まります。次のオブジェクトで watch メソッドを呼び出せます
Mongo::Client: 配置内のすべてのデータベースにわたるすべてのコレクションに対する変更を監視します。ただし、システム コレクションまたはadmin、 、 データベースのコレクションは対象外です。localconfigMongo::Database: 1 つのデータベース内のすべてのコレクションに対する変更を監視Mongo::Collection: 1 つのコレクションの変更を監視
次の例では、 restaurantsコレクションの変更ストリームを開き、変更が発生に応じて出力します。
stream = collection.watch stream.each do |doc| puts doc break if doc['operationType'] == 'invalidate' end
変更の監視を開始するには、上記のコードを実行します。 次に、別shellで、restaurantsコレクションを変更します。 次の例では、 nameフィールドの値が'Blarney Castle'であるドキュメントを更新します。
collection.update_one( { 'name' => 'Blarney Castle' }, { '$set' => { 'cuisine' => 'Irish' } } )
コレクションを更新すると、変更ストリームアプリケーションは変更が発生に応じて出力します。 出力される変更イベントは、次の出力のようになります。
{"_id"=>{"_data"=>"..."}, "operationType"=>"update", "clusterTime"=>#<...>, "ns"=>{"db"=>"sample_restaurants", "coll"=>"restaurants"}, "documentKey"=> {"_id"=>BSON::ObjectId('...')}, "updateDescription"=>{"updatedFields"=> {"cuisine"=>"Irish"}, "removedFields"=>[], "truncatedArrays"=>[]}}
変更ストリーム出力の変更
変更ストリーム出力を変更するには、配列内のパイプラインステージをwatchメソッドのパラメーターとして渡します。 配列には、次のステージを含めることができます。
$addFieldsまたは$set: ドキュメントに新しいフィールドを追加します$match: ドキュメントをフィルタリングします$project:ドキュメントフィールドのサブセットをプロジェクションします$replaceWithまたは$replaceRoot: 入力ドキュメントを指定されたドキュメントで置き換え$redact: ドキュメントのコンテンツを制限します$unset: ドキュメントからフィールドを削除します
次の例では、 $matchステージを含むパイプラインをwatchメソッドに渡します。 これは、アップデート操作が発生した場合にのみイベントを出力するようにwatchメソッドに指示します。
pipeline = [{ '$match' => { 'operationType' => 'update' } }] stream = collection.watch(pipeline) stream.each do |doc| puts doc break if doc['operationType'] == 'invalidate' end
監視動作の変更
watch メソッドの動作を変更するには、オプション ハッシュをパラメーターとして watch に渡します。次の表では、設定できるオプションの一部について説明しています。
オプション | 説明 |
|---|---|
| ドキュメントに加えられた変更のみを表示するのではなく、変更後にドキュメント全体を表示するかどうかを指定します。このオプションの詳細については、このガイドの「 変更前とイメージと変更後のイメージを含める 」セクションを参照してください。 |
| ドキュメントに加えられた変更のみを表示するのではなく、変更前のドキュメント全体を表示するかどうかを指定します。 このオプションの詳細については、「変更前イメージと変更後イメージを含める」を参照してください。 |
|
|
| 指定されたタイムスタンプの時点またはその後に発生した変更のみを提供するように変更ストリームに指示します。このオプションは |
| 変更ストリームカーソルに使用する照合を設定します。 |
watch オプションの完全なリストについては、 APIドキュメントの監視を参照してください。
変更前と変更後のイメージを含めます
重要
配置で MongoDB v 6.0以降が使用されている場合にのみ、コレクションで変更前と変更後のイメージを有効にできます。
デフォルトでは 、コレクションに対して操作を実行すると、対応する変更イベントには操作の前後で変更されたフィールドとその値のみが含まれます。
watch変更されたフィールドに加えて、ドキュメントの 変更前のイメージ 、変更前のドキュメントの完全なバージョンを返すよう、 メソッドに指示できます。変更ストリームイベントに変更前のイメージを含めるには、full_document_before_change オプションを設定するオプション ハッシュを watch に渡します。このオプションは、次の string 値に設定できます。
'whenAvailable': 変更イベントには、変更イベント用に変更されたドキュメントの変更前のイメージが含まれます。 変更前のイメージが利用できない場合、この変更イベントフィールドの値はnilになります。'required': 変更イベントには、変更イベント用に変更されたドキュメントの変更前のイメージが含まれます。 変更前のイメージが利用できない場合、サーバーはエラーを発生させます。'off':（デフォルト）変更イベントには、 変更されたドキュメントの変更前のイメージは含まれません。
また、変更されたwatch フィールドに加えて、ドキュメントの 変更後のイメージ 、変更後のドキュメントの完全なバージョンを返すよう、 メソッドに指示することもできます。変更ストリームイベントに変更後のイメージを含めるには、full_document オプションを設定するオプション ハッシュを watch に渡します。このオプションは、次の string 値に設定できます。
'updateLookup': 変更イベントには、変更後一定時間の変更されたドキュメント全体のコピーが含まれます。'whenAvailable': 変更イベントには、変更イベントの変更されたドキュメントの変更後のイメージが含まれます。 変更後イメージが利用できない場合、この変更イベントフィールドの値はnilになります。'required': 変更イベントには、変更イベントの変更されたドキュメントの変更後のイメージが含まれます。 変更後のイメージが利用できない場合、サーバーはエラーを発生させます。'default':（デフォルト）変更イベントには、変更されたドキュメントの変更後のイメージは含まれません。
次の例では、コレクションでwatchメソッドを呼び出し、 full_documentオプションを設定して更新されたドキュメントの変更後のイメージを含めます。
options = { full_document: 'updateLookup' } stream = collection.watch([], options) stream.each do |doc| puts doc break if doc['operationType'] == 'invalidate' end
別shellで実行中されている変更ストリームアプリケーションでは、前述の更新例を使用して restaurantsコレクション内のドキュメントを更新すると、次の出力のような変更イベントが出力されます。
{"_id"=>{"_data"=>"..."}, "operationType"=>"update", "clusterTime"=> #<...1>, "wallTime"=>..., "fullDocument"=>{"_id"=>BSON::ObjectId('...'), "address"=>{"building"=>"202-24", "coord"=>[-73.9250442, 40.5595462], "street"=>"Rockaway Point Boulevard", "zipcode"=>"11697"}, "borough"=>"Queens", "cuisine"=>"Irish", "grades"=>[...], "name"=>"Blarney Castle", "restaurant_id"=>"40366356"}, "ns"=> {"db"=>"sample_restaurants", "coll"=>"restaurants"}, "documentKey"=> {"_id"=>BSON::ObjectId('...')}, "updateDescription"=>{"updatedFields"=> {"cuisine"=>"Irish"}, "removedFields"=>[], "truncatedArrays"=>[]}}
Tip
変更前と変更後のイメージの詳細については、Change Streams MongoDB Serverマニュアルの「 とドキュメントの変更 前イメージおよび変更後イメージ 」を参照してください。
詳細情報
Change Streams変更ストリームの詳細については、MongoDB Server マニュアルの 「 ストリーム」 を参照してください。
API ドキュメント
このガイドで説明したメソッドや型の詳細については、次の API ドキュメントを参照してください。