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.

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.

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.

cs, err := coll.Watch(ctx, mongo.Pipeline{})
require.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 examples below assume that you have connected to a MongoDB replica set and have accessed a database that contains an inventory collection.

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

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

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

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

$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.

cursor = db.inventory.watch()
document = 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.

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.

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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)
require.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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

MongoClient mongoClient = new MongoClient( new MongoClientURI("mongodb://host1:port1,host2:port2..."));

// 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 any operations where the username is alice, or operations where the operationType 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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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 (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供）

• $redact

• $set (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供）

• $unset (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供）

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

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

$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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

pipeline = [
    {"$match": {"fullDocument.username": "alice"}},
    {"$addFields": {"newField": "this is an added field!"}},
]
cursor = db.inventory.watch(pipeline=pipeline)
document = 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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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 (Available starting in MongoDB 4.2)

• $redact

• $set (Available starting in MongoDB 4.2)

• $unset (Available starting in MongoDB 4.2)

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"选项传递给mongoc_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.

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.

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))
require.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.

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.

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

To return the most current majority-committed version of the updated document, pass full_document='updateLookup' to the db.collection.watch() method.

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.

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.在下面的示例中，所有更新操作通知都包含一个fullDocument字段，该字段表示受更新操作影响的文档的当前版本。

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

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.

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.

$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.

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 = next(cursor)

To return the most current majority-committed version of the updated document, pass full_document: 'updateLookup' to the db.watch() method.

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.

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.

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