public function watch(array $pipeline = [], array $options = [])
    {
        if ( ! isset($options['readPreference'])) {
            $options['readPreference'] = $this->readPreference;
        }

        $server = $this->manager->selectServer($options['readPreference']);

        /* Although change streams require a newer version of the server than
         * read concerns, perform the usual wire version check before inheriting
         * the collection's read concern. In the event that the server is too
         * old, this makes it more likely that users will encounter an error
         * related to change streams being unsupported instead of an
         * UnsupportedException regarding use of the "readConcern" option from
         * the Aggregate operation class. */
        if ( ! isset($options['readConcern']) && \MongoDB\server_supports_feature($server, self::$wireVersionForReadConcern) && ! $this->isInTransaction($options)) {
            $options['readConcern'] = $this->readConcern;
        }

        if ( ! isset($options['typeMap'])) {
            $options['typeMap'] = $this->typeMap;
        }

        $operation = new Watch($this->manager, $this->databaseName, $this->collectionName, $pipeline, $options);

        return $operation->execute($server);
    }