@param string $source
@param array  $pipeline
@param array  $options

@return array
@throws \ManaPHP\Mongodb\Exception
163605
public function aggregate($source, $pipeline, $options = [])
    {
        if ($pos = strpos($source, '.')) {
            $db = substr($source, 0, $pos);
            $collection = substr($source, $pos + 1);
        } else {
            $db = $this->_default_db;
            $collection = $source;
        }

        try {
            $command = ['aggregate' => $collection, 'pipeline' => $pipeline];
            if ($options) {
                $command = array_merge($command, $options);
            }
            if (!isset($command['cursor'])) {
                $command['cursor'] = ['batchSize' => 1000];
            }
            return $this->command($command, $db);
        } catch (RuntimeException $e) {
            throw new MongodbException([
                '`:aggregate` aggregate for `:collection` collection failed: :msg',
                'aggregate' => json_encode($pipeline),
                'collection' => $source,
                'msg' => $e->getMessage()
            ]);
        }
    }
231873
public static function create($source)
    {
        if (is_array($source)) {
            $source = new Collection($source);
        }

        return parent::create($source);
    }
1174860
public function aggregate($collectionName, array $pipeline) {
		$collection = $this->getCollection($collectionName);
		
		if ($this->_profilerEnabled) {
			$message = 'db.' . $collectionName . '.aggregate(' . json_encode($pipeline) . ');';
			$event   = $this->getProfiler()->startRead($message);
		}

		$result  = $collection->aggregate($pipeline);
		$success = (bool) $result['ok'];
		
		if ($this->_profilerEnabled) {
			$success
				? $event->stopSuccess()
				: $event->stopError();
		}
		
		if (! $success) {
			throw new Exceptions\OperationError('Aggregation error: "' . $result['errmsg'] . '"', $result['code']);
		}
		
		if (! empty($result['result'])) {
			return $result['result'];
		}
	}
1353266
public function img($pipeline, $src, array $options = [])
    {
        return \yii\helpers\Html::img($this->url($pipeline, $src), $options);
    }
389349
public function pipeAll(ReadableStreamInterface $source, WritableStreamInterface $destination, array $options = [])
    {
        // TODO: use stream_copy_to_stream
        // it is 4x faster than this
        // but can lose data under load with no way to recover it
        $destination->emit('pipe', array($source));

        $source->on('data', function ($data) use ($source, $destination) {
            $feedMore = $destination->write($data);
            if (false === $feedMore) {
                $source->pause();
            }
        });

        $destination->on('drain', function () use ($source) {
            $source->resume();
        });

        $end = isset($options['end']) ? $options['end'] : true;

        if ($end && $source !== $destination) {
            $source->on('end', function () use ($destination) {
                $destination->end();
            });
        }

        return $destination;
    }
481160
protected function createBaseBuilder(array $options = [])
    {
        $query = $this->createQueryBuilder();

        $this->applyResourceOptions($query, $options);

        if (empty($options['sort'])) {
            $this->defaultSort($query, $options);
        }

        return $query;
    }
373644
public function create($query, array $options = []) {
		$this->_checkConnection();

		$defaults = [
			'w' => $this->_config['w'],
			'wTimeoutMS' => $this->_config['wTimeoutMS'],
			'fsync' => false
		];
		$options += $defaults;

		$params = compact('query', 'options');

		return Filters::run($this, __FUNCTION__, $params, function($params) {
			$exporter = $this->_classes['exporter'];
			$prefix = $this->_config['gridPrefix'];

			$query   = $params['query'];
			$options = $params['options'];

			$args    = $query->export($this, ['keys' => ['source', 'data']]);
			$data    = $exporter::get('create', $args['data']);
			$source  = $args['source'];

			if ($source === "{$prefix}.files" && isset($data['create']['file'])) {
				$result = ['ok' => true];
				$data['create']['_id'] = $this->_saveFile($data['create']);
			} else {
				$result = $this->connection->{$source}->insert($data['create'], $options);
				$result = $this->_ok($result);
			}

			if ($result === true || isset($result['ok']) && (boolean) $result['ok'] === true) {
				if ($query->entity()) {
					$query->entity()->sync($data['create']['_id']);
				}
				return true;
			}
			return false;
		});
	}
962899
public function query($query, $options = [])
    {
        $url = $this->instanceURLRepo->get();
        $url .= $this->resourceRepo->get('query');
        $url .= '?q=';
        $url .= urlencode($query);

        $queryResults = $this->request($url, $options);

        return $queryResults;
    }
928465
public function getData(Options $options = null)
    {
        if ($options === null) {
            $options = $this->getOptions();
        }

        $select = $this->assignOptions(clone $this->getSelect(), $options);

        $sql = new Sql($this->adapter);

        $sql_string = $sql->buildSqlString($select);
        //echo $this->select->getSqlString($this->adapter->getPlatform());
        //echo "----" . var_dump($sql_string) . "----\n";
        // In ZF 2.3.0 an empty query will return SELECT .*
        // In ZF 2.4.0 and empty query will return SELECT *
        if (in_array($sql_string, ['', 'SELECT .*', 'SELECT *'], true)) {
            throw new Exception\EmptyQueryException(__METHOD__ . ': Cannot return data of an empty query');
        }
        $this->query_string = $sql_string;

        try {
            $results = $this->adapter->query($sql_string, Adapter::QUERY_MODE_EXECUTE);
            //$stmt = $sql->prepareStatementForSqlObject( $select );
            //$results = $stmt->execute();
            //var_dump(get_class($results));

            $r = new ResultSet($results);
            $r->setSource($this);
            $r->setHydrationOptions($options->getHydrationOptions());

            if ($options->hasLimit() && $options->getLimit() > 0) {
                //$row = $this->adapter->query('select FOUND_ROWS() as total_count')->execute()->current();
                $row = $this->adapter->createStatement('select FOUND_ROWS() as total_count')->execute()->current();
                $r->setTotalRows($row['total_count']);
            } else {
                $r->setTotalRows($r->count());
            }
        } catch (\Exception $e) {
            throw new Exception\ErrorException(__METHOD__ . ': Cannot retrieve data (' . $e->getMessage() . ')');
        }

        return $r;
    }
1008076
public static function serve($source, $destination, $options = [])
    {
        $options = array_merge(self::$defaultOptions, $options);

        try {
            ServeConvertedWebP::serve($source, $destination, $options);
        } catch (\Exception $e) {
            self::performFailAction(
                $options['fail'],
                $options['fail-when-original-unavailable'],
                $source,
                $destination,
                $options
            );
        }
    }
194552
public function getCollection(array $options = array())
    {
        /** @var ClientEntity[] $collection */
        $collection = $this->getRepository()->findAll();
        $resource   = new Resource($this->generateBrowseUrl(), array('count' => count($collection)));

        foreach ($collection as $element) {
            $resource->setEmbedded('client', $this->createResourceFromClient($element));
        }

        return $resource;
    }
588866
public function pipe($object, $pipeline = null)
    {
        $pipeline = $pipeline ?: 'default';

        return call_user_func(
            $this->pipelines[$pipeline], new Pipeline($this->container), $object
        );
    }
621320
public function load($source, array $options = []): self
    {
        $loader = $this->getLoader();
        $data = $loader->load($source, $options);

        expect_type($data, stdClass::class);

        static::mergeObjects($this, $data);

        return $this;
    }
1235333
public function buildPipeline(PipelineInterface $pipeline)
    {
        if ($pipeline->getType() !== PipelineInterface::TYPE_BUCKET_SELECTOR) {
            throw new \InvalidArgumentException("Query builder : invalid aggregation type {$pipeline->getType()}.");
        }

        $aggParams = [
            'buckets_path'  => $pipeline->getBucketsPath(),
            'script'        => $pipeline->getScript(),
            'gap_policy'    => $pipeline->getGapPolicy(),
        ];

        return ['bucket_selector' => $aggParams];
    }
217091
public function execute(array $options = []) : Iterator
    {
        // Force cursor to be used
        $options = array_merge($options, ['cursor' => true]);

        $cursor = $this->collection->aggregate($this->getPipeline(), $options);
        assert($cursor instanceof Cursor);

        return $this->prepareIterator($cursor);
    }
47034
public function parse($source, array $options = null) {
		$result = simplexml_load_string($source);
		
		if ($result === false) {
			throw new InvalidContent('Unable to parse the XML content');
		}
		
		return $this->simpleXmlToArray($result);
	}
1237249
protected function _fetchCmd(
        Horde_Imap_Client_Interaction_Pipeline $pipeline,
        $options
    )
    {
        $fetch = new Horde_Imap_Client_Data_Format_List();
        $sequence = $options['ids']->sequence;

        /* Build an IMAP4rev1 compliant FETCH query. We handle the following
         * criteria:
         *   BINARY[.PEEK][<section #>]<<partial>> (RFC 3516)
         *     see BODY[] response
         *   BINARY.SIZE[<section #>] (RFC 3516)
         *   BODY[.PEEK][<section>]<<partial>>
         *     <section> = HEADER, HEADER.FIELDS, HEADER.FIELDS.NOT, MIME,
         *                 TEXT, empty
         *     <<partial>> = 0.# (# of bytes)
         *   BODYSTRUCTURE
         *   ENVELOPE
         *   FLAGS
         *   INTERNALDATE
         *   MODSEQ (RFC 7162)
         *   RFC822.SIZE
         *   UID
         *
         * No need to support these (can be built from other queries):
         * ===========================================================
         *   ALL macro => (FLAGS INTERNALDATE RFC822.SIZE ENVELOPE)
         *   BODY => Use BODYSTRUCTURE instead
         *   FAST macro => (FLAGS INTERNALDATE RFC822.SIZE)
         *   FULL macro => (FLAGS INTERNALDATE RFC822.SIZE ENVELOPE BODY)
         *   RFC822 => BODY[]
         *   RFC822.HEADER => BODY[HEADER]
         *   RFC822.TEXT => BODY[TEXT]
         */

        foreach ($options['_query'] as $type => $c_val) {
            switch ($type) {
            case Horde_Imap_Client::FETCH_STRUCTURE:
                $fetch->add('BODYSTRUCTURE');
                break;

            case Horde_Imap_Client::FETCH_FULLMSG:
                if (empty($c_val['peek'])) {
                    $this->openMailbox($this->_selected, Horde_Imap_Client::OPEN_READWRITE);
                }
                $fetch->add(
                    'BODY' .
                    (!empty($c_val['peek']) ? '.PEEK' : '') .
                    '[]' .
                    $this->_partialAtom($c_val)
                );
                break;

            case Horde_Imap_Client::FETCH_HEADERTEXT:
            case Horde_Imap_Client::FETCH_BODYTEXT:
            case Horde_Imap_Client::FETCH_MIMEHEADER:
            case Horde_Imap_Client::FETCH_BODYPART:
            case Horde_Imap_Client::FETCH_HEADERS:
                foreach ($c_val as $key => $val) {
                    $cmd = ($key == 0)
                        ? ''
                        : $key . '.';
                    $main_cmd = 'BODY';

                    switch ($type) {
                    case Horde_Imap_Client::FETCH_HEADERTEXT:
                        $cmd .= 'HEADER';
                        break;

                    case Horde_Imap_Client::FETCH_BODYTEXT:
                        $cmd .= 'TEXT';
                        break;

                    case Horde_Imap_Client::FETCH_MIMEHEADER:
                        $cmd .= 'MIME';
                        break;

                    case Horde_Imap_Client::FETCH_BODYPART:
                        // Remove the last dot from the string.
                        $cmd = substr($cmd, 0, -1);

                        if (!empty($val['decode']) &&
                            $this->_capability('BINARY')) {
                            $main_cmd = 'BINARY';
                            $pipeline->data['binaryquery'][$key] = $val;
                        }
                        break;

                    case Horde_Imap_Client::FETCH_HEADERS:
                        $cmd .= 'HEADER.FIELDS';
                        if (!empty($val['notsearch'])) {
                            $cmd .= '.NOT';
                        }
                        $cmd .= ' (' . implode(' ', array_map('Horde_String::upper', $val['headers'])) . ')';

                        // Maintain a command -> label lookup so we can put
                        // the results in the proper location.
                        $pipeline->data['fetch_lookup'][$cmd] = $key;
                    }

                    if (empty($val['peek'])) {
                        $this->openMailbox($this->_selected, Horde_Imap_Client::OPEN_READWRITE);
                    }

                    $fetch->add(
                        $main_cmd .
                        (!empty($val['peek']) ? '.PEEK' : '') .
                        '[' . $cmd . ']' .
                        $this->_partialAtom($val)
                    );
                }
                break;

            case Horde_Imap_Client::FETCH_BODYPARTSIZE:
                if ($this->_capability('BINARY')) {
                    foreach ($c_val as $val) {
                        $fetch->add('BINARY.SIZE[' . $val . ']');
                    }
                }
                break;

            case Horde_Imap_Client::FETCH_ENVELOPE:
                $fetch->add('ENVELOPE');
                break;

            case Horde_Imap_Client::FETCH_FLAGS:
                $fetch->add('FLAGS');
                break;

            case Horde_Imap_Client::FETCH_IMAPDATE:
                $fetch->add('INTERNALDATE');
                break;

            case Horde_Imap_Client::FETCH_SIZE:
                $fetch->add('RFC822.SIZE');
                break;

            case Horde_Imap_Client::FETCH_UID:
                /* A UID FETCH will always return UID information (RFC 3501
                 * [6.4.8]). Don't add to query as it just creates a longer
                 * FETCH command. */
                if ($sequence) {
                    $fetch->add('UID');
                }
                break;

            case Horde_Imap_Client::FETCH_SEQ:
                /* Nothing we need to add to fetch request unless sequence is
                 * the only criteria (see below). */
                break;

            case Horde_Imap_Client::FETCH_MODSEQ:
                /* The 'changedsince' modifier implicitly adds the MODSEQ
                 * FETCH item (RFC 7162 [3.1.4.1]). Don't add to query as it
                 * just creates a longer FETCH command. */
                if (empty($options['changedsince'])) {
                    $fetch->add('MODSEQ');
                }
                break;
            }
        }

        /* If empty fetch, add UID to make command valid. */
        if (!count($fetch)) {
            $fetch->add('UID');
        }

        /* Add changedsince parameters. */
        if (empty($options['changedsince'])) {
            $fetch_cmd = $fetch;
        } else {
            /* We might just want the list of UIDs changed since a given
             * modseq. In that case, we don't have any other FETCH attributes,
             * but RFC 3501 requires at least one specified attribute. */
            $fetch_cmd = array(
                $fetch,
                new Horde_Imap_Client_Data_Format_List(array(
                    'CHANGEDSINCE',
                    new Horde_Imap_Client_Data_Format_Number($options['changedsince'])
                ))
            );
        }

        /* The FETCH command should be the only command issued by this library
         * that should ever approach the command length limit.
         * @todo Move this check to a more centralized location (_command()?).
         * For simplification, assume that the UID list is the limiting factor
         * and split this list at a sequence comma delimiter if it exceeds
         * the character limit. */
        foreach ($options['ids']->split($this->_capability()->cmdlength) as $val) {
            $cmd = $this->_command(
                $sequence ? 'FETCH' : 'UID FETCH'
            )->add(array(
                $val,
                $fetch_cmd
            ));
            $pipeline->add($cmd);
        }
    }