Supervisar los cambios en los datos

Datos de muestra

Los ejemplos de esta guía utilizan el sample_restaurants.restaurants Colección de los conjuntos de datos de muestra de Atlas. Para aprender a crear un clúster gratuito de MongoDB Atlas y cargar los conjuntos de datos de muestra, consulte Introducción al controlador .NET/C#.

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 eventos de actualización del monitor

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

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

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

Ejemplo de división de eventos de cambio grandes

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

Este ejemplo indica al controlador que detecte cambios y divida eventos de cambio que superen el límite de 16 MB. El código imprime el documento de cambio para cada evento e invoca métodos auxiliares para reensamblar los fragmentos de evento:

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

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

El ejemplo anterior utiliza los métodos GetNextChangeStreamEvent(), GetNextChangeStreamEventAsync() y MergeFragment() para reensamblar 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 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; 
        }
    }
}

// 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;
        }
    }
}

Intercalación

Para configurar la intercalación para tu operación, crea una instancia de la clase intercalación.

La siguiente tabla describe los parámetros que acepta el constructor Collation. También enumera la propiedad de clase correspondiente que se puede usar para leer el valor de cada configuración.

Más información sobre la intercalación en la página Intercalación en el manual de MongoDB Server.

Incluir imágenes previas y posteriores

De forma predeterminada, al realizar una operación en una colección, el evento de cambio correspondiente incluye solo el delta de los campos modificados por dicha 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. A continuación, pase 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, configure la FullDocumentBeforeChange opción con uno de los siguientes valores:

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

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

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

• ChangeStreamFullDocumentOption.UpdateLookup:El evento de cambio incluye una copia de todo el documento modificado desde algún tiempo después del cambio.

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

• ChangeStreamFullDocumentOption.RequiredEl evento de cambio incluye una imagen posterior del documento modificado. Si la imagen posterior no está disponible, el controlador genera un error.

El siguiente ejemplo abre un flujo de cambios en una colección e incluye la imagen posterior de los documentos actualizados mediante la opción FullDocument. Seleccione la pestaña Synchronous o Asynchronous 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 = collection.Watch(pipeline, options))
{
    foreach (var change in cursor.ToEnumerable())
    {
        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 = await collection.WatchAsync(pipeline, options);
await cursor.ForEachAsync(change =>
{
    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 imágenes previas y posteriores, consulte Flujos de cambio con imágenes previas y posteriores 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:

• Mirar()

• WatchAsync()

• Opciones de cambio de flujo

• UpdateOne()