Supervise los datos con Change Streams

Datos de muestra

Los ejemplos en esta guía utilizan lo siguiente Course struct como modelo para documentos en la colección courses:

type Course struct {
	Title      string
	Enrollment int32
}

Para ejecutar los ejemplos de esta guía, cargue estos documentos en la colección courses de la base de datos db utilizando el siguiente snippet:

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

Cada documento contiene una descripción de una asignatura universitaria que incluye el título y la cantidad máxima de inscritos, correspondientes a los campos title y enrollment en cada documento.

Ejemplo

El siguiente ejemplo abre un flujo de cambios en la colección courses e imprime los eventos de flujo de cambios a medida que ocurren:

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

Si modificas la colección courses en un programa o shell aparte, este código imprime tus cambios a medida que se producen. Insertar un documento con un valor title de "Advanced Screenwriting" y un valor de enrollment de 20 produce el siguiente evento de cambio:

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 un ejemplo completamente ejecutable, consulta la sección Abrir un ejemplo de Change Stream: Archivo completo de esta guía.

Eventos de cambio de filtro

Utiliza el parámetro de pipeline para modificar la salida del flujo de cambios. Este parámetro te permite observar solo ciertos eventos de cambio. Da formato al parámetro de pipeline como un arreglo de documentos, donde cada documento representa una etapa de agregación.

Puedes utilizar las siguientes etapas de pipeline en este parámetro:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

El siguiente ejemplo abre un flujo de cambios en la base de datos db, pero solo observa nuevas operaciones de eliminación:

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

Ejemplo de apertura de un Change Stream: Archivo completo

El siguiente ejemplo abre un flujo de cambios en la colección restaurants e imprime los documentos insertados:

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

Mire un ejemplo totalmente ejecutable.

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

Pre y publicación imágenes

Cuando realices cualquier operación CRUD en una colección, por defecto, el documento de evento de cambio correspondiente solo contiene el delta de los campos modificados por la operación. Puedes ver el documento completo antes y después de un cambio, además del delta, especificando la configuración en el parámetro options del método Watch().

Si deseas ver la imagen posterior de un documento, la versión completa del documento después de un cambio, configura el campo FullDocument del parámetro options en uno de los siguientes valores:

• UpdateLookup: El documento del evento de cambio incluye una copia de todo el documento modificado.

• WhenAvailableEl documento de evento de cambio incluye una post-imagen del documento modificado para eventos de cambio si la post-imagen está disponible.

• RequiredLa salida es la misma que para WhenAvailable, pero el controlador genera un error del lado del servidor si la imagen posterior no está disponible.

Si deseas ver la pre-imagen de un documento, la versión completa del documento antes de un cambio, establece el campo FullDocumentBeforeChange del parámetro options en uno de los siguientes valores:

• WhenAvailableEl documento de evento de cambio incluye una preimagen del documento modificado para eventos de cambio si la preimagen está disponible.

• Required: La salida es la misma que para WhenAvailable, pero el controlador genera un error del lado del servidor si la preimagen no está disponible.

Ejemplo

El siguiente ejemplo llama al método Watch() en la colección courses. Especifica un valor para el campo FullDocument del parámetro options para generar una copia de todo el documento modificado, en lugar de solo los campos modificados:

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

Actualizar el valor de enrollment del documento con el title de "World Fiction" de 35 a 30 produce el siguiente evento de cambio:

{"_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": []}}

Si no se especifica la opción FullDocument, la misma operación de actualización ya no mostrará el valor de "fullDocument" en el documento del evento de cambio.

Documentación de la API

Para aprender más sobre el método Watch(), consulte la siguiente documentación de la API:

• Watch() para colecciones

• Watch() para bases de datos

• Watch() para los clientes