MongoDB\Collection::watch()

定义

MongoDB\Collection::watch()

对集合执行变更流操作。 可以监视变更流以了解集合级别的变更。

function watch(
    array $pipeline = [],
    array $options = []
): MongoDB\ChangeStream

参数

$pipeline : array|object

要附加到初始$changeStream阶段的阶段管道。

$options : array

指定所需选项的数组。

名称	类型	说明
batchSize	整型	指定游标的批处理大小，这将应用于初始aggregate命令和任何后续getMore命令。 这决定了服务器每次响应中返回的变更事件的最大数量。 无论batchSize选项如何，change stream的初始aggregate命令响应通常不包含任何文档，除非使用另一个选项来配置其起点（例如 startAfter ）。
编解码器	MongoDB\Codec\DocumentCodec	用于对文档进行编码或解码的编解码器。 此选项与typeMap选项互斥。 默认为集合的编解码器。默认 codec 选项的继承优先于 typeMap 选项的继承。 1.17 版本中的新增功能。
排序规则	数组|对象	排序规则允许用户为string比较指定特定于语言的规则，例如字母大小写和重音符号规则。 指定排序规则时， locale字段为必填字段；所有其他排序规则字段都是可选的。 有关字段的说明，请参阅排序规则文档。 从 MongoDB 4.2 开始，如果省略，则默认为简单的二进制比较。 在早期版本中，单个collection上打开的change stream将继承该collection的默认排序规则。
comment	混合	使用户可以指定任意注释来帮助通过数据库分析器、currentOp 输出和日志来跟踪操作。 评论可以是自 MongoDB 4.4 以来任何有效的 BSON 类型。早期服务器版本仅支持字符串值。 1.13 版本中的新增功能。
完整文档	字符串	确定如何为更新操作填充fullDocument响应字段。 默认，change stream仅返回更新操作的字段增量（通过updateDescription字段），并省略fullDocument 。插入和替换操作始终包含fullDocument字段。删除操作会省略该字段，因为文档不再存在。 指定“updateLookup”以返回更新文档的当前多数提交版本。 如果collection启用了changeStreamPreAndPostImages ，MongoDB 6.0+ 允许返回修改后文档的帖子图像。指定“whenAvailable”以在帖子可用时返回该帖子，如果不可用则返回 null 值。指定“必需”以在可用时返回后图像，否则引发错误。 支持以下值： MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP MongoDB\Operation\Watch::FULL_DOCUMENT_WHEN_AVAILABLE MongoDB\Operation\Watch::FULL_DOCUMENT_REQUIRED 这是$changeStream管道阶段的一个选项。
fullDocumentBeforeChange	字符串	确定如何填充fullDocumentBeforeChange响应字段。默认情况下，省略该字段。 如果collection启用了changeStreamPreAndPostImages ，MongoDB 6.0+ 允许返回已修改文档的前像。指定“whenAvailable”以在前像可用时返回前像，如果不可用则返回空值。 指定“required”以在可用时返回前像，否则引发错误。 支持以下值： MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_WHEN_AVAILABLE MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_REQUIRED 这是$changeStream管道阶段的一个选项。 1.13 版本中的新增功能。
maxAwaitTimeMS	整型	正整数，表示服务器在没有可用数据时阻塞 getMore 操作的时间限制（以毫秒为单位）。
事务外的	MongoDB\Driver\ReadConcern	读取关注以用于操作。 默认为集合的读关注。
readPreference	MongoDB\Driver\ReadPreference	读取要用于操作的偏好。 默认为集合的读取偏好。 这用于初始change stream聚合和自动恢复期间的服务器选择。
resumeAfter	数组|对象	指定新change stream的逻辑点。此处可使用change stream返回的文档中的_id字段。 将该选项与startAfter和/或startAtOperationTime一起使用会导致服务器错误。 这些选项是互斥的。 这是$changeStream管道阶段的一个选项。
会话	MongoDB\Driver\Session	与操作相关联的客户端会话。
showExpandedEvents	布尔	如果为 true，则指示服务器在 change stream 中包含其他事件。可能包含的其他事件包括： createIndexes dropIndexes modify create shardCollection reshardCollection （服务器 6.1+） refineCollectionShardKey （服务器 6.1+） 6.0 之前的服务器版本不支持此功能，如果使用，将导致执行时出现异常。 这是$changeStream管道阶段的一个选项。 1.13 版本中的新增功能。
startAfter	数组|对象	指定新change stream的逻辑点。此处可使用变更流返回的文档中的_id字段。 与resumeAfter不同，此选项可与“invalidate”事件中的恢复令牌一起使用。 将该选项与resumeAfter和/或startAtOperationTime一起使用会导致服务器错误。 这些选项是互斥的。 4.2 之前的服务器版本不支持此功能，如果使用，将导致执行时出现异常。 这是$changeStream管道阶段的一个选项。 1.5 版本中的新增功能。
startAtOperationTime	MongoDB\BSON\TimestampInterface	如果指定，则变更流将仅提供在指定时间戳时或之后发生的更改。 来自 MongoDB 4.0+ 服务器的命令响应包含可在此处使用的operationTime 。 默认情况下，将使用初始aggregate命令返回的operationTime （如果可用）。 将该选项与resumeAfter和/或startAfter一起使用会导致服务器错误。 这些选项是互斥的。 4.0 之前的服务器版本不支持此功能，如果使用，将导致执行时出现异常。 这是$changeStream管道阶段的一个选项。
typeMap	阵列	类型映射 应用于游标，决定如何将 BSON 文档转换为 PHP 值。默认为集合的类型映射。

Return Values

一个MongoDB\ChangeStream 对象，允许通过 迭代器迭代 变更流中的事件 接口。

错误/异常

MongoDB\Exception\UnexpectedValueException ，如果来自服务器的命令响应格式错误。

MongoDB\Exception\UnsupportedException，如果所选服务器使用了选项但不支持选项（例如collation、readConcern、writeConcern）。

MongoDB\Exception\InvalidArgumentException 用于与参数或选项解析相关的错误。

MongoDB\ 驱动程序\Exception\RuntimeException 对于扩展级别的其他错误（例如连接错误）。

示例

此示例在迭代change stream时报告事件。

<?php
$uri = 'mongodb://rs1.example.com,rs2.example.com/?replicaSet=myReplicaSet';
$collection = (new MongoDB\Client($uri))->test->inventory;
$changeStream = $collection->watch();
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    if ($event['operationType'] === 'invalidate') {
        break;
    }
    $ns = sprintf('%s.%s', $event['ns']['db'], $event['ns']['coll']);
    $id = json_encode($event['documentKey']['_id']);
    switch ($event['operationType']) {
        case 'delete':
            printf("Deleted document in %s with _id: %s\n\n", $ns, $id);
            break;
        case 'insert':
            printf("Inserted new document in %s\n", $ns);
            echo json_encode($event['fullDocument']), "\n\n";
            break;
        case 'replace':
            printf("Replaced new document in %s with _id: %s\n", $ns, $id);
            echo json_encode($event['fullDocument']), "\n\n";
            break;
        case 'update':
            printf("Updated document in %s with _id: %s\n", $ns, $id);
            echo json_encode($event['updateDescription']), "\n\n";
            break;
    }
}

假设在上述脚本迭代change stream时插入、更新和删除了文档，则输出将类似于以下内容：

Inserted new document in test.user
{"_id":{"$oid":"5b329c4874083047cc05e60a"},"username":"bob"}
Inserted new document in test.products
{"_id":{"$oid":"5b329c4d74083047cc05e60b"},"name":"Widget","quantity":5}
Updated document in test.user with _id: {"$oid":"5b329a4f74083047cc05e603"}
{"updatedFields":{"username":"robert"},"removedFields":[]}

另请参阅

• MongoDB\Client::watch()

• MongoDB\Database::watch()

• MongoDB 手册中的聚合管道文档

• 手册中的Change Streams 文档MongoDB

• MongoDB 手册中的变更事件文档