Monitorar alterações de dados

Dados de amostra

Os exemplos neste guia usam a sample_restaurants.restaurants coleção dos conjuntos de dados de amostra do Atlas . Para saber como criar um cluster do MongoDB Atlas gratuito e carregar os conjuntos de dados de amostra, consulte a seção início rápido.

Os exemplos desta página utilizam as seguintes classes Restaurant, Address e 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; }
}

Exemplo de eventos de atualização de monitoramento

O exemplo a seguir usa o parâmetro pipeline para abrir um fluxo de alterações que registra somente operações de atualização. Selecione a aba Asynchronous ou Synchronous para ver o código correspondente.

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

Exemplo de divisão de grandes eventos de mudança

Se o seu aplicação gerar eventos de alteração que excedam 16 MB, o servidor retornará um erro BSONObjectTooLarge. Para evitar esse erro, você pode usar o estágio de pipeline $changeStreamSplitLargeEvent para divisão os eventos em fragmentos menores. A API de agregação do driver .NET /C# inclui o método ChangeStreamSplitLargeEvent(), que você pode usar para adicionar o estágio $changeStreamSplitLargeEvent ao pipeline do change stream.

Este exemplo instrui o driver a observar as alterações e divisão os eventos de alteração que excedam o limite de 16 MB. O código imprime o documento de alteração para cada evento e chama métodos assistente para reagrupar quaisquer fragmentos 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);
}

O exemplo anterior utiliza os métodos GetNextChangeStreamEvent(), GetNextChangeStreamEventAsync() e MergeFragment() para reagrupar fragmentos de evento de alteração em um único change stream documento. O seguinte código define estes 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 pré-imagens e pós-imagens

Por padrão, quando você executa uma operação em uma collection, o evento de alteração correspondente inclui somente o delta dos campos modificados por essa operação. Para ver o documento completo antes ou depois de uma alteração, crie um objeto ChangeStreamOptions e especifique as opções FullDocumentBeforeChange ou FullDocument . Em seguida, passe o objeto ChangeStreamOptions para o método Watch() ou WatchAsync() .

A pré-imagem é a versão completa de um documento antes de uma alteração. Para incluir a pré-imagem no evento de fluxo de alteração, defina a opção FullDocumentBeforeChange para um dos seguintes valores:

• ChangeStreamFullDocumentBeforeChangeOption.WhenAvailable: o evento de alteração inclui uma pré-imagem do documento modificado para eventos de alteração somente se a pré-imagem estiver disponível.

• ChangeStreamFullDocumentBeforeChangeOption.Required: o evento de alteração inclui uma pré-imagem do documento modificado para eventos de alteração. Se a pré-imagem não estiver disponível, o driver gerará um erro.

A pós-imagem é a versão completa de um documento após uma alteração. Para incluir a pós-imagem na alteração de evento de fluxo, defina a opção FullDocument para um dos seguintes valores:

• ChangeStreamFullDocumentOption.UpdateLookup: o evento de alteração inclui uma cópia de todo o documento alterado de algum tempo após a alteração.

• ChangeStreamFullDocumentOption.WhenAvailable: O evento de alteração inclui uma pós-imagem do documento modificado para eventos de alteração somente se a pós-imagem estiver disponível.

• ChangeStreamFullDocumentOption.Required: o evento de alteração inclui uma pós-imagem do documento modificado para eventos de alteração. Se a pós-imagem não estiver disponível, o driver gerará um erro.

O exemplo a seguir abre um fluxo de alteração em uma collection e inclui a pós-imagem de documentos atualizados especificando a opção FullDocument . Selecione a aba Asynchronous ou Synchronous para ver o código correspondente.

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

Executar o exemplo de código anterior e atualizar um documento que tenha um valor de resulta na seguinte saída do "name" "Blarney Castle" :

{ "_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 saber mais sobre pré e pós-imagens, consulte Change Streams com pré e pós-imagens de documentos no manual do MongoDB Server .

Documentação da API

Para saber mais sobre qualquer um dos métodos ou tipos discutidos neste guia, consulte a seguinte documentação da API:

• watch()

• WatchAsync()

• ChangeStreamOptions

• UpdateOne()