Supervisar los cambios en los datos

Datos de muestra

Los ejemplos en esta guía utilizan la colección sample_restaurants.restaurants de los conjuntos de datos de muestra de Atlas. Para aprender a crear un clúster de MongoDB Atlas gratuito y cargar los conjuntos de datos de muestra, consulta el Inicio rápido.

En los ejemplos de esta página se utilizan las siguientes clases Restaurant, Address y GradeEntry como modelos:

public class Restaurant
{
    public ObjectId Id { get; set; }
    public string Name { get; set; }
    [BsonElement("restaurant_id")]
    public string RestaurantId { get; set; }
    public string Cuisine { get; set; }
    public Address Address { get; set; }
    public string Borough { get; set; }
    public List<GradeEntry> Grades { get; set; }
}

public class Address
{
    public string Building { get; set; }
    [BsonElement("coord")]
    public double[] Coordinates { get; set; }
    public string Street { get; set; }
    [BsonElement("zipcode")]
    public string ZipCode { get; set; }
}

public class GradeEntry
{
    public DateTime Date { get; set; }
    public string Grade { get; set; }
    public float? Score { get; set; }
}

Ejemplo de monitorización de eventos de actualización

El siguiente ejemplo utiliza el parámetro pipeline para abrir un flujo de cambios que registre solo las operaciones de actualización. Selecciona la pestaña Asynchronous o Synchronous para ver el código correspondiente.

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change stream and prints the changes as they're received
using (var cursor = await collection.WatchAsync(pipeline))
{
    await cursor.ForEachAsync(change =>
    {
        Console.WriteLine("Received the following change: " + change);
    });
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change streams and print the changes as they're received
using (var cursor = collection.Watch(pipeline))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following change: " + change);
    }
}

Ejemplo de división de grandes eventos de cambios

Si la aplicación genera eventos de cambio que superan los 16 MB de tamaño, el servidor devuelve un error BSONObjectTooLarge. Para evitar este error, puedes usar la etapa $changeStreamSplitLargeEvent pipeline para dividir los eventos en fragmentos más pequeños. La API de agregación del driver .NET/C# incluye el método ChangeStreamSplitLargeEvent(), que puedes usar para añadir la etapa $changeStreamSplitLargeEvent a la pipeline de flujo de cambios.

Este ejemplo instruye al driver a observar los cambios y dividir los eventos de cambio que superen el límite de 16 MB. El código imprime el documento de cambio para cada evento y llama a métodos asistentes para reensamblar cualquier fragmento de evento:

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = await collection.WatchAsync(pipeline);
await foreach (var completeEvent in GetNextChangeStreamEventAsync(cursor))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = collection.Watch(pipeline);
foreach (var completeEvent in GetNextChangeStreamEvent(cursor.ToEnumerable().GetEnumerator()))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

El ejemplo anterior utiliza los métodos GetNextChangeStreamEvent(), GetNextChangeStreamEventAsync() y MergeFragment() para reagrupar los fragmentos de eventos de cambio en un único documento de flujo de cambios. El siguiente código define estos métodos:

// Fetches the next complete change stream event
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    var changeStreamEnumerator = GetNextChangeStreamEventFragmentAsync(changeStreamCursor).GetAsyncEnumerator();
    while (await changeStreamEnumerator.MoveNextAsync())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                await changeStreamEnumerator.MoveNextAsync();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventFragmentAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    while (await changeStreamCursor.MoveNextAsync())
    {
        foreach (var changeStreamEvent in changeStreamCursor.Current)
        {
            yield return changeStreamEvent;
        }
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value;
        }
    }
}

// Fetches the next complete change stream event
private static IEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEvent<TDocument>(
IEnumerator<ChangeStreamDocument<TDocument>> changeStreamEnumerator)
{
    while (changeStreamEnumerator.MoveNext())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                changeStreamEnumerator.MoveNext();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value; 
        }
    }
}

Incluir imágenes preoperativas y postoperatorias

De forma predeterminada, cuando realizas una operación en una colección, el evento de cambio correspondiente solo incluye el delta de los campos modificados por esa operación. Para ver el documento completo antes o después de un cambio, cree un objeto ChangeStreamOptions y especifique las opciones FullDocumentBeforeChange o FullDocument. Luego, pasa el objeto ChangeStreamOptions al método Watch() o WatchAsync().

La preimagen es la versión completa de un documento antes de un cambio. Para incluir la preimagen en el evento de flujo de cambios, se debe establecer la opción FullDocumentBeforeChange en uno de los siguientes valores:

• ChangeStreamFullDocumentBeforeChangeOption.WhenAvailable: El evento de cambio incluye una preimagen del documento modificado para los eventos de cambio solo si la preimagen está disponible.

• ChangeStreamFullDocumentBeforeChangeOption.RequiredEl evento de cambio incluye una preimagen del documento modificado para los eventos de cambio. Si la preimagen no está disponible, el driver genera un error.

La post-imagen es la versión completa de un documento después de un cambio. Para incluir la post-imagen en el evento del flujo de cambios, configure la opción FullDocument en uno de los siguientes valores:

• ChangeStreamFullDocumentOption.UpdateLookup: El evento de cambio incluye una copia de todo el documento cambiado a partir de cierto tiempo después del cambio.

• ChangeStreamFullDocumentOption.WhenAvailable: El evento de cambio incluye una postimagen del documento modificado para eventos de cambio solo si la postimagen está disponible.

• ChangeStreamFullDocumentOption.Required: El evento de cambio incluye una imagen posterior del documento modificado para eventos de cambio. Si la imagen posterior no está disponible, el driver genera un error.

El siguiente ejemplo abre un flujo de cambios en una colección e incluye la posimagen de los documentos actualizados al especificar la opción FullDocument. Selecciona la pestaña Asynchronous o Synchronous para ver el código correspondiente.

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using var cursor = await collection.WatchAsync(pipeline, options);
await cursor.ForEachAsync(change =>
{
    Console.WriteLine(change.FullDocument.ToBsonDocument());
});

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using (var cursor = collection.Watch(pipeline, options))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine(change.FullDocument.ToBsonDocument());
    }
}

Ejecutar el ejemplo de código anterior y actualizar un documento que tenga un valor de "name" igual a "Blarney Castle" resulta en la siguiente salida del flujo de cambios:

{ "_id" : ObjectId("..."), "name" : "Blarney Castle", "restaurant_id" : "40366356",
"cuisine" : "Traditional Irish", "address" : { "building" : "202-24", "coord" : [-73.925044200000002, 40.5595462],
"street" : "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens", "grades" : [...] }

Para obtener más información sobre pre-imágenes y post-imágenes, consulta Change Streams con preimágenes y postimágenes de documentos en el manual de MongoDB Server.

Documentación de la API

Para aprender más sobre cualquiera de los métodos o tipos analizados en esta guía, consulta la siguiente documentación de API:

• Watch()

• WatchAsync()

• ChangeStreamOptions

• UpdateOne()