MongoDB\Collection::watch()

定义

MongoDB\Collection::watch()

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

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

参数

$pipeline : array|object

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

$options : array

指定所需选项的数组。

名称	类型	说明
batchSize	整型	变更流中返回的每个批处理中的最大文档数，适用于 aggregate 命令。默认下，aggregate 命令的初始批处理大小为 101 个文档，后续每个批处理的最大大小为 16 兆字节 (MiB)。此选项可以实施小于 16 MiB 的限制，但不能强制执行大于 MiB 的限制。如果将 batchSize设立为导致批量大于 16 MiB 的限制，则此选项无效。 无论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 手册中的变更事件文档