Abrir Change Streams

Overview

En esta guía, puedes aprender a usar un flujo de cambios para supervisar los cambios en tiempo real de tus datos. Un flujo de cambios es una funcionalidad del MongoDB Server que permite a tu aplicación suscribirse a los cambios de datos en una única 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 detectores de movimiento

• Cree una aplicación que supervise los cambios en los precios de las acciones

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

Puedes especificar un conjunto de operadores de agregación para filtrar y transformar los datos que recibe tu aplicación. Cuando te conectes a una implementación de MongoDB v6.0 o posterior, también puedes 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, navega a las siguientes secciones:

• Datos de muestra

• Abre un flujo de cambios

• Aplicar operadores de agregación al flujo de cambios

• Incluir imágenes preoperativas y postoperatorias

• Información Adicional

Datos de muestra

Los ejemplos de esta guía supervisan 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

Puedes abrir un flujo de cambios para suscribirte a tipos específicos de cambios de datos y producir eventos de cambios en tu aplicación.

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

El tipo de estructura sobre el que llames el watch() método determina el alcance de los eventos para los que el flujo de cambios escucha. 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 al flujo de cambios

Se puede encadenar el método pipeline() al método watch() para especificar qué eventos de cambio recibe el flujo de cambios. Pase un pipeline de agregación como un parámetro a pipeline().

Para aprender qué operadores de agregación admite tu versión de MongoDB Server, consulta Modificar salida de 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 preoperativas y postoperatorias

Puedes 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 preimagen o una postimagen, debes realizar las siguientes acciones:

• Habilita las preimágenes y las postimágenes para la colección en tu implementación de MongoDB.

  Tip

  Para obtener instrucciones sobre cómo habilitar imágenes pre/post en la implementación, consulte Change Streams con imágenes pre/post en documentos en el manual del servidor.

  Para aprender a indicar al controlador que cree una colección con preimágenes y postimágenes habilitadas, consulta la sección Crea una colección con preimágenes y postimágenes habilitadas de esta página.

• Configura tu flujo de cambios para recuperar o bien las preimágenes, las postimágenes o ambas. Durante esta configuración, se puede indicar al controlador que requiera imágenes previas y posteriores o incluirlas 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 tu flujo de cambios para grabar la posimagen en los eventos de cambio, consulta el Ejemplo de Configuración de la Posimagen 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