Change Streams更改流

The C examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的C示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;

collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

The C# examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的C#示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

The Go examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的Go示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

cs, err := coll.Watch(ctx, mongo.Pipeline{})
assert.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

The Java examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的Java示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

The Kotlin examples below assume that you are connected to a MongoDB replica set and can access a database that contains the inventory collection. 下面的Kotlin示例假设您连接到MongoDB副本集，并且可以访问包含inventory集合的数据库。To learn more about completing these tasks, see the Kotlin Coroutine Driver Databases and Collections guide.要了解有关完成这些任务的更多信息，请参阅Kotlin Coroutine驱动程序数据库和集合指南。

val job = launch {
    val changeStream = collection.watch()
    changeStream.collect {
        println("Received a change event: $it")
    }
}

The examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

cursor = db.inventory.watch()
document = await cursor.next()

The Node.js examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的Node.js示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

The following example uses stream to process the change events.以下示例使用流来处理更改事件。

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream
  .on('change', next => {
    // process next document
  })
  .once('error', () => {
    // handle error
  });

Alternatively, you can also use iterator to process the change events:或者，您还可以使用迭代器来处理更改事件：

const collection = db.collection('inventory');
const changeStream = collection.watch();
const next = await changeStream.next();

ChangeStream extends EventEmitter.ChangeStream扩展了EventEmitter。

The examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

$changeStream = $db->inventory->watch();
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

The Python examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的Python示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

cursor = db.inventory.watch()
next(cursor)

The examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的示例假设您已连接到MongoDB副本集，并访问了包含inventory集合的数据库。

cursor = inventory.watch.to_enum
next_change = cursor.next

The Swift (Async) examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.

let inventory = db.collection("inventory")

// Option 1: retrieve next document via next()
let next = inventory.watch().flatMap { cursor in
    cursor.next()
}

// Option 2: register a callback to execute for each document
let result = inventory.watch().flatMap { cursor in
    cursor.forEach { event in
        // process event
        print(event)
    }
}

The Swift (Sync) examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.下面的Swift(Sync)示例假设您已连接到MongoDB副本集并访问了包含inventory集合的数据库。

let inventory = db.collection("inventory")
let changeStream = try inventory.watch()
let next = changeStream.next()

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

pipeline = BCON_NEW ("pipeline",
                     "[",
                     "{",
                     "$match",
                     "{",
                     "fullDocument.username",
                     BCON_UTF8 ("alice"),
                     "}",
                     "}",
                     "{",
                     "$addFields",
                     "{",
                     "newField",
                     BCON_UTF8 ("this is an added field!"),
                     "}",
                     "}",
                     "]");

stream = mongoc_collection_watch (collection, pipeline, &opts);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change =>
        change.FullDocument["username"] == "alice" ||
        change.OperationType == ChangeStreamOperationType.Delete)
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        "{ $addFields : { newField : 'this is an added field!' } }");

var collection = database.GetCollection<BsonDocument>("inventory");
using (var cursor = collection.Watch(pipeline))
{
    while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
    var next = cursor.Current.First();
}

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"$or",
	bson.A{
		bson.D{{"fullDocument.username", "alice"}},
		bson.D{{"operationType", "delete"}}}}},
}}}
cs, err := coll.Watch(ctx, pipeline)
assert.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

MongoClient mongoClient = MongoClients.create("mongodb://<username>:<password>@<host>:<port>");

// Select the MongoDB database and collection to open the change stream against

MongoDatabase db = mongoClient.getDatabase("myTargetDatabase");

MongoCollection<Document> collection = db.getCollection("myTargetCollection");

// Create $match pipeline stage.
List<Bson> pipeline = singletonList(Aggregates.match(Filters.or(
   Document.parse("{'fullDocument.username': 'alice'}"),
   Filters.in("operationType", asList("delete")))));

// Create the change stream cursor, passing the pipeline to the
// collection.watch() method

MongoCursor<Document> cursor = collection.watch(pipeline).iterator();

The pipeline list includes a single $match stage that filters for any operations that meet one or both of the following criteria:

• username value is alice
• operationType value is delete

Passing the pipeline to the watch() method directs the change stream to return notifications after passing them through the specified pipeline.

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

val pipeline = listOf(
    Aggregates.match(
        or(
            eq("fullDocument.username", "alice"),
            `in`("operationType", listOf("delete"))
        )
))

val job = launch {
    val changeStream = collection.watch(pipeline)
    changeStream.collect {
        println("Received a change event: $it")
    }
}

The pipeline list includes a single $match stage that filters for any operations that meet one or both of the following criteria:pipeline列表包含一个$match阶段，用于筛选满足以下一个或两个条件的任何操作：

• username value is 值是alice
• operationType value is 值是delete

Passing the pipeline to the watch() method directs the change stream to return notifications after passing them through the specified pipeline.将pipeline传递给watch()方法会指示更改流在通过指定管道后返回通知。

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

pipeline = [
    {"$match": {"fullDocument.username": "alice"}},
    {"$addFields": {"newField": "this is an added field!"}},
]
cursor = db.inventory.watch(pipeline=pipeline)
document = await cursor.next()

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

The following example uses stream to process the change events.以下示例使用流来处理更改事件。

const pipeline = [
  { $match: { 'fullDocument.username': 'alice' } },
  { $addFields: { newField: 'this is an added field!' } }
];

const collection = db.collection('inventory');
const changeStream = collection.watch(pipeline);
changeStream
  .on('change', next => {
    // process next document
  })
  .once('error', error => {
    // handle error
  });

Alternatively, you can also use iterator to process the change events:或者，您还可以使用迭代器来处理更改事件：

const changeStreamIterator = collection.watch(pipeline);
const next = await changeStreamIterator.next();

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

$pipeline = [
    ['$match' => ['fullDocument.username' => 'alice']],
    ['$addFields' => ['newField' => 'this is an added field!']],
];
$changeStream = $db->inventory->watch($pipeline);
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

pipeline = [
    {"$match": {"fullDocument.username": "alice"}},
    {"$addFields": {"newField": "this is an added field!"}},
]
cursor = db.inventory.watch(pipeline=pipeline)
next(cursor)

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

let pipeline: [BSONDocument] = [
    ["$match": ["fullDocument.username": "alice"]],
    ["$addFields": ["newField": "this is an added field!"]]
]
let inventory = db.collection("inventory")

// Option 1: use next() to iterate
let next = inventory.watch(pipeline, withEventType: BSONDocument.self).flatMap { changeStream in
    changeStream.next()
}

// Option 2: register a callback to execute for each document
let result = inventory.watch(pipeline, withEventType: BSONDocument.self).flatMap { changeStream in
    changeStream.forEach { event in
        // process event
        print(event)
    }
}

You can control change stream output by providing an array of one or more of the following pipeline stages when configuring the change stream:在配置更改流时，您可以通过提供以下一个或多个管道阶段的数组来控制更改流输出：

• $addFields
• $match
• $project
• $replaceRoot
• $replaceWith
• $redact
• $set
• $unset

let pipeline: [BSONDocument] = [
    ["$match": ["fullDocument.username": "alice"]],
    ["$addFields": ["newField": "this is an added field!"]]
]
let inventory = db.collection("inventory")
let changeStream = try inventory.watch(pipeline, withEventType: BSONDocument.self)
let next = changeStream.next()

To return the most current majority-committed version of the updated document, pass the "fullDocument" option with the "updateLookup" value to the mongoc_collection_watch method.要返回更新文档的最新多数提交版本，请将带有"updateLookup"值的"fullDocument"选项传递给mongo_collection_watch方法。

In the example below, all update operations notifications include a fullDocument field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个fullDocument字段，表示受更新操作影响的文档的当前版本。

BSON_APPEND_UTF8 (&opts, "fullDocument", "updateLookup");
stream = mongoc_collection_watch (collection, pipeline, &opts);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

To return the most current majority-committed version of the updated document, pass "FullDocument = ChangeStreamFullDocumentOption.UpdateLookup" to the db.collection.watch() method.要返回更新文档的最新多数提交版本，请将"FullDocument = ChangeStreamFullDocumentOption.UpdateLookup"传递给db.collection.watch()方法。

In the example below, all update operations notifications include a FullDocument field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个FullDocument字段，该字段表示受更新操作影响的文档的当前版本。

var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
var cursor = inventory.Watch(options);
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

To return the most current majority-committed version of the updated document, SetFullDocument(options.UpdateLookup) change stream option.要返回更新文档的最新多数提交版本，请使用SetFullDocument(options.UpdateLookup)更改流选项。

cs, err := coll.Watch(ctx, mongo.Pipeline{}, options.ChangeStream().SetFullDocument(options.UpdateLookup))
assert.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

To return the most current majority-committed version of the updated document, pass FullDocument.UPDATE_LOOKUP to the db.collection.watch.fullDocument() method.要返回更新文档的最新多数提交版本，请将FullDocument.UPDATE_LOOKUP传递给db.collection.watch.fullDocument()方法。

In the example below, all update operations notifications include a FullDocument field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个FullDocument字段，该字段表示受更新操作影响的文档的当前版本。

cursor = inventory.watch().fullDocument(FullDocument.UPDATE_LOOKUP).iterator();
next = cursor.next();

To return the most current majority-committed version of the updated document, pass FullDocument.UPDATE_LOOKUP to the ChangeStreamFlow.fullDocument() method.要返回更新文档的最新多数提交版本，请将FullDocument.UPDATE_LOOKUP传递给ChangeStreamFlow.fullDocument()方法。

In the example below, all update operations notifications include a FullDocument field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个FullDocument字段，该字段表示受更新操作影响的文档的当前版本。

val job = launch {
    val changeStream = collection.watch()
        .fullDocument(FullDocument.UPDATE_LOOKUP)
    changeStream.collect {
        println(it)
    }
}

To return the most current majority-committed version of the updated document, pass full_document='updateLookup' to the db.collection.watch() method.要返回更新文档的最新多数提交版本，请将full_document='updateLookup'传递给db.collection.watch()方法。

In the example below, all update operations notifications include a `full_document field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个full_document字段，表示受更新操作影响的文档的当前版本。

cursor = db.inventory.watch(full_document="updateLookup")
document = await cursor.next()

To return the most current majority-committed version of the updated document, pass { fullDocument: 'updateLookup' } to the db.collection.watch() method.要返回更新文档的最新多数提交版本，请将{ fullDocument: 'updateLookup' }传递给db.collection.watch()方法。

In the example below, all update operations notifications include a fullDocument field that represents the current version of the document affected by the update operation.

The following example uses stream to process the change events.以下示例使用流来处理更改事件。

const collection = db.collection('inventory');
const changeStream = collection.watch([], { fullDocument: 'updateLookup' });
changeStream
  .on('change', next => {
    // process next document
  })
  .once('error', error => {
    // handle error
  });

Alternatively, you can also use iterator to process the change events:或者，您还可以使用迭代器来处理更改事件：

const changeStreamIterator = collection.watch([], { fullDocument: 'updateLookup' });
const next = await changeStreamIterator.next();

To return the most current majority-committed version of the updated document, pass "fullDocument' => \MongoDB\Operation\ChangeStreamCommand::FULL_DOCUMENT_UPDATE_LOOKUP" to the db.watch() method.要返回更新文档的最新多数提交版本，请将"fullDocument' => \MongoDB\Operation\ChangeStreamCommand::FULL_DOCUMENT_UPDATE_LOOKUP"传递给db.watch()方法。

In the example below, all update operations notifications include a fullDocument field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个fullDocument字段，表示受更新操作影响的文档的当前版本。

$changeStream = $db->inventory->watch([], ['fullDocument' => \MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP]);
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

To return the most current majority-committed version of the updated document, pass full_document='updateLookup' to the db.collection.watch() method.要返回更新文档的最新多数提交版本，请将full_document='updateLookup'传递给db.collection.watch()方法。

In the example below, all update operations notifications include a full_document field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个full_document字段，该字段表示受更新操作影响的文档的当前版本。

cursor = db.inventory.watch(full_document="updateLookup")
next(cursor)

To return the most current majority-committed version of the updated document, pass full_document: 'updateLookup' to the db.watch() method.要返回更新文档的最新多数提交版本，请将full_document: 'updateLookup'传递给db.watch()方法。

In the example below, all update operations notifications include a full_document field that represents the current version of the document affected by the update operation.在下面的示例中，所有更新操作通知都包含一个full_document字段，该字段表示受更新操作影响的文档的当前版本。

cursor = inventory.watch([], full_document: 'updateLookup').to_enum
next_change = cursor.next

To return the most current majority-committed version of the updated document, pass options: ChangeStreamOptions(fullDocument: .updateLookup) to the watch() method.要返回更新文档的最新多数提交版本，请将options: ChangeStreamOptions(fullDocument: .updateLookup)传递给watch()方法。

let inventory = db.collection("inventory")

// Option 1: use next() to iterate
let next = inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
    .flatMap { changeStream in
        changeStream.next()
    }

// Option 2: register a callback to execute for each document
let result = inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
    .flatMap { changeStream in
        changeStream.forEach { event in
            // process event
            print(event)
        }
    }

To return the most current majority-committed version of the updated document, pass options: ChangeStreamOptions(fullDocument: .updateLookup) to the watch() method.要返回更新文档的最新多数提交版本，请将options: ChangeStreamOptions(fullDocument: .updateLookup)传递给watch()方法。

let inventory = db.collection("inventory")
let changeStream = try inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
let next = changeStream.next()