オープンChange Streams

Overview

このガイドでは、変更ストリームを使用してデータに対するリアルタイムの変更を監視する方法を学習できます。 変更ストリームは、アプリケーションが単一のコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。

変更ストリームを開くと、次のアクションの実行に役立ちます。

• 動作を検出するデバイスなど、イベントの追跡とReactを有効にする

• 株価の変化を監視するアプリケーションを作成する

• 特定のジョブ 位置の従業員のアクティビティのログを生成

アプリケーションが受信したデータをフィルタリングして変換するための集計演算子のセットを指定できます。 MongoDB 配置 v6.0 以降に接続すると、変更前と変更後のドキュメント データを含むようにイベントを構成することもできます。

変更ストリームを開いて構成する方法については、次のセクションに移動します。

• サンプル データ
• 変更ストリームを開く
• 変更ストリームへの集計演算子の適用
• 変更前と変更後のイメージを含めます
• 詳細情報

サンプル データ

このガイドの例では、 directorsコレクションの変更を監視します。 このコレクションには、 構造体としてモデル化された次のドキュメントが含まれているとします。

let docs = vec! [
    Director {
        name: "Todd Haynes".to_string(),
        movies: vec! ["Far From Heaven".to_string(), "Carol".to_string()],
        oscar_noms: 1,
    },
    Director {
        name: "Jane Campion".to_string(),
        movies: vec! ["The Piano".to_string(), "Bright Star".to_string()],
        oscar_noms: 5,
    }
];

変更ストリームを開く

変更ストリームを開いて、特定のタイプのデータ変更をサブスクライブし、アプリケーション内で変更イベントを生成できます。

変更ストリームを開くには、 Collection 、 Database 、またはClientインスタンスでwatch()メソッドを呼び出します。

watch()メソッドを呼び出す構造体によって、変更ストリームがリッスンするイベントの範囲が決まります。 次の表は、呼び出しオブジェクトに基づくwatch()メソッドの動作を説明したものです。

let mut change_stream = my_coll.watch().await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Document: {:?}", event.full_document);
}

Operation performed: Insert
Document: Some(Director { name: "Wes Anderson", movies: [...], oscar_noms: 7 })

変更ストリームへの集計演算子の適用

pipeline()メソッドとwatch()メソッドを連鎖させて、変更ストリームが受信する変更イベントを指定できます。 集計パイプラインをパラメータとしてpipeline()に渡します。

MongoDB Server のバージョンがサポートする集計演算子については、サーバー マニュアルの「変更ストリーム出力の変更 」を参照してください。

let mut update_change_stream = my_coll.watch()
    .pipeline(vec![doc! { "$match" : doc! { "operationType" : "update" } }])
    .await?;
while let Some(event) = update_change_stream.next().await.transpose()? {
    println!("Update performed: {:?}", event.update_description);
}

Update performed: Some(UpdateDescription { updated_fields: Document({
"oscar_noms": Int64(2)}), removed_fields: [], truncated_arrays: Some([]) })

変更前と変更後のイメージを含めます

変更イベントは、以下のデータを含めるか省略するように構成できます。

• 変更前のイメージ: 操作前のドキュメントのバージョンを表すドキュメント（存在する場合）

• 変更後イメージ: 操作後のドキュメントのバージョンを表すドキュメント（存在する場合）

変更前イメージまたは変更後イメージを含む変更ストリーム イベントを受信するには、次のアクションを実行する必要があります。

• MongoDB 配置のコレクションの変更前と変更後のイメージを有効にします。

  Tip

  配置で変更前と変更後のイメージを有効にする方法については、サーバー マニュアルの「 Change Streams とドキュメントの変更前イメージおよび変更後イメージ 」を参照してください。

  変更前イメージと変更後イメージが有効になっているコレクションを作成するようにドライバーに指示する方法については、このページの「 変更前イメージと変更後イメージが有効になっているコレクションの作成」セクションを参照してください。

• 変更ストリームを設定して、変更前のイメージと変更後のイメージのどちらかを取得します。 この構成中に、変更前と変更後のイメージを要求するようにドライバーに指示したり、利用可能な場合にのみ含めるようにドライバーに指示したりできます。

  Tip

  変更イベントに変更前のイメージを記録するように変更ストリームを構成する方法について詳しくは、このページの「 変更前のイメージの構成例」を参照してください。

  変更イベントで変更後のイメージを記録するように変更ストリームを構成するには、このページの「 変更後のイメージの構成例」を参照してください。

let enable = ChangeStreamPreAndPostImages::builder().enabled(true).build();
let result = my_db.create_collection("directors")
    .change_stream_pre_and_post_images(enable)
    .await?;

let pre_image = FullDocumentBeforeChangeType::Required;
let mut change_stream = my_coll.watch()
    .full_document_before_change(pre_image)
    .await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Pre-image: {:?}", event.full_document_before_change);
}

Operation performed: Update
Pre-image: Some(Director { name: "Jane Campion", movies: ["The Piano",
"Bright Star"], oscar_noms: 5 })

let post_image = FullDocumentType::WhenAvailable;
let mut change_stream = my_coll.watch()
    .full_document(post_image)
    .await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Post-image: {:?}", event.full_document);
}

Operation performed: Delete
Post-image: None

詳細情報