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版本新增。
排序规则	数组|对象	Collation allows users to specify language-specific rules for string comparison, such as rules for lettercase and accent marks. When specifying collation, the locale field is mandatory; all other collation fields are optional. For descriptions of the fields, see Collation Document. 如果省略，此选项默认为简单的二进制比较。
comment	混合	使用户可以指定任意注释来帮助通过数据库分析器、currentOp 输出和日志来跟踪操作。 评论可以是从MongoDB 4.4 开始的任何有效BSON类型。早期服务器版本仅支持字符串值。 1.13版本新增。
完整文档	字符串	确定如何为更新操作填充fullDocument响应字段。 默认，change stream仅返回更新操作的字段增量（通过updateDescription字段），并省略fullDocument 。插入和替换操作始终包含fullDocument字段。删除操作会省略该字段，因为文档不再存在。 指定“updateLookup”以返回更新文档的当前多数提交版本。 从MongoDB 6.0 开始，如果集合启用了 changeStreamPreAndPostImages，则可以访问权限修改后文档的后像。指定 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	Read concern to use for the operation. Defaults to the collection's read concern.
readPreference	MongoDB\Driver\ReadPreference	Read preference to use for the operation. Defaults to the collection's read preference. 这用于初始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一起使用会导致服务器错误。 这些选项是互斥的。 这是$changeStream管道阶段的一个选项。 1.5版本新增。
startAtOperationTime	MongoDB\BSON\TimestampInterface	如果指定，变更流将仅提供在指定时间戳时或之后发生的更改。默认下，将使用初始 aggregate 命令返回的 operationTime（如果可用）。 将该选项与resumeAfter和/或startAfter一起使用会导致服务器错误。 这些选项是互斥的。 要使用此选项，您必须连接到MongoDB 6.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 手册中的变更事件文档