protected function doAggregateCursor(array $pipeline, array $options = array())
    {
        if ( ! method_exists('RiakCollection', 'aggregateCursor')) {
            throw new BadMethodCallException('RiakCollection::aggregateCursor() is not available');
        }

        list($commandOptions, $clientOptions) = isset($options['socketTimeoutMS']) || isset($options['timeout'])
            ? $this->splitCommandAndClientOptions($options)
            : array($options, array());

        if (is_scalar($commandOptions['cursor'])) {
            unset($commandOptions['cursor']);
        }

        $timeout = isset($clientOptions['socketTimeoutMS'])
            ? $clientOptions['socketTimeoutMS']
            : (isset($clientOptions['timeout']) ? $clientOptions['timeout'] : null);

        $riakCollection = $this->riakCollection;
        $commandCursor = $this->retry(function() use ($riakCollection, $pipeline, $commandOptions) {
            return $riakCollection->aggregateCursor($pipeline, $commandOptions);
        });

        $commandCursor = $this->wrapCommandCursor($commandCursor);

        if (isset($timeout)) {
            $commandCursor->timeout($timeout);
        }

        return $commandCursor;
    }