Flujos de cambio abiertos

Overview

En esta guía, aprenderá a usar un flujo de cambios para supervisar los cambios en sus datos en tiempo real. Un flujo de cambios es una función de MongoDB Server que permite que su aplicación se suscriba a los cambios en los datos de una sola colección, base de datos o implementación.

Puedes abrir un flujo de cambios para ayudar a realizar las siguientes acciones:

• Permitir que los dispositivos rastreen y reaccionen a eventos, como cámaras con detección de movimiento.

• Crear una aplicación que monitoree los cambios en los precios de las acciones

• Generar un registro de la actividad de los empleados para puestos de trabajo específicos

Puede especificar un conjunto de operadores de agregación para filtrar y transformar los datos que recibe su aplicación. Al conectarse a una implementación de MongoDB v6.0 o posterior, también puede configurar los eventos para que incluyan los datos del documento antes y después del cambio.

Para aprender a abrir y configurar un flujo de cambios, navegue a las siguientes secciones:

• Datos de muestra

• Abre un flujo de cambios

• Aplicar operadores de agregación a su flujo de cambios

• Incluir imágenes previas y posteriores

• Información Adicional

Datos de muestra

Los ejemplos de esta guía controlan los cambios en el directors Colección. Supongamos que esta colección contiene los siguientes documentos, modelados como estructuras:

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,
    }
];

Abre un flujo de cambios

Puede abrir un flujo de cambios para suscribirse a tipos específicos de cambios de datos y producir eventos de cambio en su aplicación.

Para abrir un flujo de cambios, llame al método watch() en una instancia Collection, Database o Client.

El tipo de estructura en el que se llama al método watch() determina el alcance de los eventos que escucha el flujo de cambios. La siguiente tabla describe el comportamiento del método watch() según el objeto que lo llama:

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 })

Aplicar operadores de agregación a su flujo de cambios

Puede encadenar el método pipeline() con el método watch() para especificar qué eventos de cambio recibe el flujo de cambios. Pase una canalización de agregación como parámetro a pipeline().

Para saber qué operadores de agregación admite su versión de MongoDB Server, consulte Modificar la salida del flujo de cambios en el manual del servidor.

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([]) })

Incluir imágenes previas y posteriores

Puede configurar el evento de cambio para que contenga u omita los siguientes datos:

• Preimagen: un documento que representa la versión del documento antes de la operación, si existe

• Post-imagen: un documento que representa la versión del documento después de la operación, si existe

Para recibir eventos de flujo de cambios que incluyan una imagen previa o posterior, debe realizar las siguientes acciones:

• Habilite imágenes previas y posteriores para la colección en su implementación de MongoDB.

  Tip

  Para saber cómo habilitar imágenes previas y posteriores en su implementación, consulte Flujos de cambio con imágenes previas y posteriores de documentos en el manual del servidor.

  Para saber cómo indicarle al controlador que cree una colección con imágenes previas y posteriores habilitadas, consulte la sección Crear una colección con imágenes previas y posteriores habilitadas de esta página.

• Configure su flujo de cambios para recuperar las imágenes previas y posteriores, o ambas. Durante esta configuración, puede indicar al controlador que solicite imágenes previas y posteriores o que las incluya solo cuando estén disponibles.

  Tip

  Para configurar su flujo de cambios para registrar la imagen previa en eventos de cambio, consulte el Ejemplo de configuración de imagen previa en esta página.

  Para configurar su flujo de cambios para registrar la imagen posterior en eventos de cambio, consulte el Ejemplo de configuración de imagen posterior en esta página.

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

Información Adicional