Monitore dados com change streams

Dados de amostra

Os exemplos nesta aba usam o seguinte struct Course como um modelo para documentos na coleção courses:

type Course struct {
	Title      string
	Enrollment int32
}

Para executar os exemplos neste guia, carregue estes documentos na coleção courses no banco de dados do db utilizando o seguinte trecho:

coll := client.Database("db").Collection("courses")
docs := []interface{}{
	Course{Title: "World Fiction", Enrollment: 35},
	Course{Title: "Abstract Algebra", Enrollment: 60},
}
result, err := coll.InsertMany(context.TODO(), docs)

Cada documento contém uma descrição de um curso universitário que inclui o nome do curso e o número máximo de matrículas, correspondentes aos campos title e enrollment em cada documento.

Exemplo

O exemplo a seguir abre um fluxo de alteração na coleção courses e imprime os eventos de fluxo de alteração conforme ocorrem:

changeStream, err := coll.Watch(context.TODO(), mongo.Pipeline{})
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
// Iterates over the cursor to print the change stream events
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

Se você modificar a coleção courses em um programa ou shell separado, este código imprimirá as alterações conforme elas ocorrerem. Inserir um documento com um valor de title de "Advanced Screenwriting" e um valor de enrollment de 20 resulta no seguinte evento de alteração:

map[_id:map[_data:...] clusterTime: {...}
documentKey:map[_id:ObjectID("...")] fullDocument:map[_id:ObjectID("...")
enrollment:20 title:Advanced Screenwriting] ns: map[coll:courses db:db]
operationType:insert]

Para ver um exemplo totalmente executável, consulte a seção Abrir um Exemplo de Fluxo de Alterações: Arquivo Completo neste guia.

Filtrar eventos de alteração

Use o parâmetro de pipeline para modificar a saída do fluxo de alteração. Esse parâmetro permite que você observe apenas determinados eventos de alteração. Formate o parâmetro pipeline como uma array de documentos, com cada documento representando um estágio de agregação.

Você pode usar os seguintes estágios de pipeline neste parâmetro:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

O exemplo seguinte abre um fluxo de alteração no banco de dados do db, mas somente assiste para novas operações de exclusão:

db := client.Database("db")
pipeline := bson.D{{"$match", bson.D{{"operationType", "delete"}}}}
changeStream, err := db.Watch(context.TODO(), mongo.Pipeline{pipeline})
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
// Iterates over the cursor to print the delete operation change events
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

Abrir um exemplo de fluxo de alteração: arquivo completo

O exemplo a seguir abre um fluxo de alteração na coleção restaurants e imprime documentos inseridos:

coll := client.Database("sample_restaurants").Collection("restaurants")
// Creates instructions to watch for insert operations
pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"operationType", "insert"}}}}}
// Creates a change stream that receives change events
cs, err := coll.Watch(context.TODO(), pipeline)
if err != nil {
	panic(err)
}
defer cs.Close(context.TODO())
fmt.Println("Waiting For Change Events. Insert something in MongoDB!")
// Prints a message each time the change stream receives an event
for cs.Next(context.TODO()) {
	var event bson.M
	if err := cs.Decode(&event); err != nil {
		panic(err)
	}
	output, err := json.MarshalIndent(event["fullDocument"], "", "    ")
	if err != nil {
		panic(err)
	}
	fmt.Printf("%s\n", output)
}
if err := cs.Err(); err != nil {
	panic(err)
}

Veja um exemplo totalmente executável.

// results truncated {
  "_id": ..., "name": "8282", "cuisine": "Korean"
}

Pré-imagens e pós-imagens

Quando você executa qualquer operação CRUD em uma coleção, por padrão, o documento de evento de alteração correspondente contém somente o delta dos campos modificados pela operação. Você pode visualizar o documento completo antes e depois de uma alteração, além do delta, especificando as configurações no parâmetro options do método Watch().

Se você quiser ver a pós-imagem de um documento, a versão completa do documento após uma alteração, defina o campo FullDocument do parâmetro options para um dos seguintes valores:

• UpdateLookup: o documento de alteração de evento inclui uma cópia de todo o documento alterado.

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

• Required: a saída é a mesma de WhenAvailable, mas o driver gera um erro no servidor se a pós-imagem não estiver disponível.

Se você quiser ver a pré-imagem de um documento, a versão completa do documento antes de uma alteração, defina o campo FullDocumentBeforeChange do parâmetro options para um dos seguintes valores:

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

• Required: a saída é a mesma de WhenAvailable, mas o driver gera um erro no servidor se a pré-imagem não estiver disponível.

Exemplo

O exemplo a seguir chama o método Watch() na coleção courses. Ele especifica um valor para o campo FullDocument do parâmetro options para gerar uma cópia de todo o documento modificado, em vez de apenas os campos alterados:

opts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
changeStream, err := coll.Watch(context.TODO(), mongo.Pipeline{}, opts)
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

Atualizar o valor de enrollment do documento com o title de "World Fiction" de 35 para 30 resulta no seguinte evento de alteração:

{"_id": {"_data": "..."},"operationType": "update","clusterTime":
{"$timestamp": {"t":"...","i":"..."}},"fullDocument": {"_id":
{"$oid":"..."},"title": "World Fiction","enrollment": {"$numberInt":"30"}},
"ns": {"db": "db","coll": "courses"},"documentKey": {"_id": {"$oid":"..."}},
"updateDescription": {"updatedFields": {"enrollment": {"$numberInt":"30"}},
"removedFields": [],"truncatedArrays": []}}

Sem especificar a opção FullDocument, a mesma operação de atualização não gera mais o valor "fullDocument" no documento de evento de alteração.

Documentação da API

Para saber mais sobre o método Watch() , consulte a seguinte documentação da API:

• Watch() para coleções

• Watch() para bancos de dados

• Watch() para clientes