symfony · fabpot · Jan 17, 2018 · Jan 10, 2018
diff --git a/src/ComposerRepository.php b/src/ComposerRepository.php
@@ -0,0 +1,62 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Flex;
+
+use Composer\Config;
+use Composer\Downloader\TransportException;
+use Composer\IO\IOInterface;
+use Composer\Repository\ComposerRepository as BaseComposerRepository;
+use Composer\Util\RemoteFilesystem;
+
+/**
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+class ComposerRepository extends BaseComposerRepository
+{
+    private $providerFiles;
+
+    protected function loadProviderListings($data)
+    {
+        if (null !== $this->providerFiles) {
+            parent::loadProviderListings($data);
+
+            return;
+        }
+
+        $data = [$data];
+
+        while ($data) {
+            $this->providerFiles = [];
+            foreach ($data as $data) {
+                $this->loadProviderListings($data);
+            }
+
+            $loadingFiles = $this->providerFiles;
+            $this->providerFiles = null;
+            $data = [];
+            $this->rfs->download($loadingFiles, function (...$args) use (&$data) {
+                $data[] = $this->fetchFile(...$args);
+            });
+        }
+    }
+
+    protected function fetchFile($filename, $cacheKey = null, $sha256 = null, $storeLastModifiedTime = false)
+    {
+        if (null !== $this->providerFiles) {
+            $this->providerFiles[] = [$filename, $cacheKey, $sha256, $storeLastModifiedTime];
+
+            return [];
+        }
+
+        return parent::fetchFile($filename, $cacheKey, $sha256, $storeLastModifiedTime);
+    }
+}
diff --git a/src/CurlDownloader.php b/src/CurlDownloader.php
@@ -0,0 +1,200 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Flex;
+
+use Composer\Downloader\TransportException;
+
+/**
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+class CurlDownloader
+{
+    private $multiHandle;
+    private $shareHandle;
+    private $jobs = [];
+    private $exceptions = [];
+
+    private static $options = [
+        'http' => [
+            'method' => CURLOPT_CUSTOMREQUEST,
+            'content' => CURLOPT_POSTFIELDS,
+            'proxy' => CURLOPT_PROXY,
+        ],
+        'ssl' => [
+            'ciphers' => CURLOPT_SSL_CIPHER_LIST,
+            'cafile' => CURLOPT_CAINFO,
+            'capath' => CURLOPT_CAPATH,
+        ],
+    ];
+
+    private static $timeInfo = [
+        'total_time' => true,
+        'namelookup_time' => true,
+        'connect_time' => true,
+        'pretransfer_time' => true,
+        'starttransfer_time' => true,
+        'redirect_time' => true,
+    ];
+
+    public function __construct()
+    {
+        $this->multiHandle = $mh = curl_multi_init();
+        curl_multi_setopt($mh, CURLMOPT_PIPELINING, /*CURLPIPE_HTTP1 | CURLPIPE_MULTIPLEX*/ 3);
+        curl_multi_setopt($mh, CURLMOPT_MAX_HOST_CONNECTIONS, 8);
+
+        $this->shareHandle = $sh = curl_share_init();
+        curl_share_setopt($sh, CURLSHOPT_SHARE, CURL_LOCK_DATA_COOKIE);
+        curl_share_setopt($sh, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS);
+        curl_share_setopt($sh, CURLSHOPT_SHARE, CURL_LOCK_DATA_SSL_SESSION);
+    }
+
+    public function get($origin, $url, $context, $file)
+    {
+        $params = stream_context_get_params($context);
+
+        $ch = curl_init();
+        $hd = fopen('php://memory', 'w+b');
+        if ($file && !$fd = @fopen($file.'~', 'w+b')) {
+            $file = null;
+        }
+        if (!$file) {
+            $fd = @fopen('php://temp/maxmemory:524288', 'w+b');
+        }
+        $headers = array_diff($params['options']['http']['header'], ['Connection: close']);
+
+        if (!isset($params['options']['http']['protocol_version'])) {
+            curl_setopt($ch, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_0);
+        } else {
+            $headers[] = 'Connection: keep-alive';
+        }
+
+        curl_setopt($ch, CURLOPT_URL, $url);
+        curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
+        curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
+        curl_setopt($ch, CURLOPT_DNS_USE_GLOBAL_CACHE, false);
+        curl_setopt($ch, CURLOPT_WRITEHEADER, $hd);
+        curl_setopt($ch, CURLOPT_FILE, $fd);
+        curl_setopt($ch, CURLOPT_SHARE, $this->shareHandle);
+
+        foreach (self::$options as $type => $options) {
+            foreach ($options as $name => $curlopt) {
+                if (isset($params['options'][$type][$name])) {
+                    curl_setopt($ch, $curlopt, $params['options'][$type][$name]);
+                }
+            }
+        }
+
+        $progress = array_diff_key(curl_getinfo($ch), self::$timeInfo);
+        $this->jobs[(int) $ch] = [
+            'progress' => $progress,
+            'ch' => $ch,
+            'callback' => $params['notification'],
+            'file' => $file,
+            'fd' => $fd,
+        ];
+
+        curl_multi_add_handle($this->multiHandle, $ch);
+        $params['notification'](STREAM_NOTIFY_RESOLVE, STREAM_NOTIFY_SEVERITY_INFO, '', 0, 0, 0, false);
+        $active = true;
+
+        try {
+            while ($active && isset($this->jobs[(int) $ch])) {
+                curl_multi_exec($this->multiHandle, $active);
+                curl_multi_select($this->multiHandle);
+
+                while ($progress = curl_multi_info_read($this->multiHandle)) {
+                    if (!isset($this->jobs[$i = (int) $h = $progress['handle']])) {
+                        continue;
+                    }
+                    $progress = array_diff_key(curl_getinfo($h), self::$timeInfo);
+                    $job = $this->jobs[$i];
+                    unset($this->jobs[$i]);
+                    curl_multi_remove_handle($this->multiHandle, $h);
+                    try {
+                        $this->onProgress($h, $job['callback'], $progress, $job['progress']);
+
+                        if ($job['file'] && CURLE_OK === curl_errno($h) && !isset($this->exceptions[$i])) {
+                            fclose($job['fd']);
+                            rename($job['file'].'~', $job['file']);
+                        }
+                    } catch (TransportException $e) {
+                        $this->exceptions[$i] = $e;
+                    }
+                }
+
+                foreach ($this->jobs as $i => $h) {
+                    if (!isset($this->jobs[$i])) {
+                        continue;
+                    }
+                    $h = $this->jobs[$i]['ch'];
+                    $progress = array_diff_key(curl_getinfo($h), self::$timeInfo);
+
+                    if ($progress !== $this->jobs[$i]['progress']) {
+                        $previousProgress = $this->jobs[$i]['progress'];
+                        $this->jobs[$i]['progress'] = $progress;
+                        try {
+                            $this->onProgress($h, $this->jobs[$i]['callback'], $progress, $previousProgress);
+                        } catch (TransportException $e) {
+                            unset($this->jobs[$i]);
+                            curl_multi_remove_handle($this->multiHandle, $h);
+                            $this->exceptions[$i] = $e;
+                        }
+                    }
+                }
+            }
+
+            if (CURLE_OK !== curl_errno($ch)) {
+                $this->exceptions[(int) $ch] = new TransportException(curl_error($ch), curl_getinfo($ch, CURLINFO_HTTP_CODE) ?: 0);
+            }
+            if (isset($this->exceptions[(int) $ch])) {
+                throw $this->exceptions[(int) $ch];
+            }
+        } finally {
+            if ($file && !isset($this->exceptions[(int) $ch])) {
+                $fd = fopen($file, 'rb');
+            }
+            unset($this->jobs[(int) $ch], $this->exceptions[(int) $ch]);
+            curl_multi_remove_handle($this->multiHandle, $ch);
+            curl_close($ch);
+
+            rewind($hd);
+            $headers = explode("\r\n", rtrim(stream_get_contents($hd)));
+            fclose($hd);
+
+            rewind($fd);
+            $contents = stream_get_contents($fd);
+            fclose($fd);
+        }
+
+        return [$headers, $contents];
+    }
+
+    private function onProgress($ch, callable $notify, array $progress, array $previousProgress)
+    {
+        if (300 <= $progress['http_code'] && $progress['http_code'] < 400) {
+            return;
+        }
+
+        if (!$previousProgress['http_code'] && $progress['http_code'] && $progress['http_code'] < 200 || 400 <= $progress['http_code']) {
+            $code = 403 === $progress['http_code'] ? STREAM_NOTIFY_AUTH_RESULT : STREAM_NOTIFY_FAILURE;
+            $notify($code, STREAM_NOTIFY_SEVERITY_ERR, curl_error($ch), $progress['http_code'], 0, 0, false);
+        }
+
+        if ($previousProgress['download_content_length'] < $progress['download_content_length']) {
+            $notify(STREAM_NOTIFY_FILE_SIZE_IS, STREAM_NOTIFY_SEVERITY_INFO, '', 0, 0, (int) $progress['download_content_length'], false);
+        }
+
+        if ($previousProgress['size_download'] < $progress['size_download']) {
+            $notify(STREAM_NOTIFY_PROGRESS, STREAM_NOTIFY_SEVERITY_INFO, '', 0, (int) $progress['size_download'], (int) $progress['download_content_length'], false);
+        }
+    }
+}
diff --git a/src/Downloader.php b/src/Downloader.php
@@ -22,7 +22,6 @@
 use Composer\Json\JsonFile;
 use Composer\Plugin\PluginEvents;
 use Composer\Plugin\PreFileDownloadEvent;
-use Composer\Util\RemoteFilesystem;
 
 /**
  * @author Fabien Potencier <fabien@symfony.com>
@@ -42,7 +41,7 @@ class Downloader
     private $flexId;
     private $eventDispatcher;
 
-    public function __construct(Composer $composer, IoInterface $io)
+    public function __construct(Composer $composer, IoInterface $io, ParallelDownloader $rfs)
     {
         if (getenv('SYMFONY_CAFILE')) {
             $this->caFile = getenv('SYMFONY_CAFILE');
@@ -56,7 +55,7 @@ public function __construct(Composer $composer, IoInterface $io)
         $this->io = $io;
         $config = $composer->getConfig();
         $this->eventDispatcher = $composer->getEventDispatcher();
-        $this->rfs = Factory::createRemoteFilesystem($io, $config);
+        $this->rfs = $rfs;
         $this->cache = new Cache($io, $config->get('cache-repo-dir').'/'.preg_replace('{[^a-z0-9.]}i', '-', $this->endpoint));
         $this->sess = bin2hex(random_bytes(16));
 
@@ -114,7 +113,7 @@ public function getRecipes(array $operations): array
                 $path .= ','.$date->format('U');
             }
             if (strlen($chunk) + strlen($path) > self::$MAX_LENGTH) {
-                $paths[] = '/p/'.$chunk;
+                $paths[] = ['/p/'.$chunk];
                 $chunk = $path;
             } elseif ($chunk) {
                 $chunk .= ';'.$path;
@@ -123,14 +122,18 @@ public function getRecipes(array $operations): array
             }
         }
         if ($chunk) {
-            $paths[] = '/p/'.$chunk;
+            $paths[] = ['/p/'.$chunk];
         }
 
-        $data = [];
-        foreach ($paths as $path) {
-            if (!$body = $this->get($path, [], false)->getBody()) {
-                continue;
+        $bodies = [];
+        $this->rfs->download($paths, function ($path) use (&$bodies) {
+            if ($body = $this->get($path, [], false)->getBody()) {
+                $bodies[] = $body;
             }
+        });
+
+        $data = [];
+        foreach ($bodies as $body) {
             foreach ($body['manifests'] as $name => $manifest) {
                 $data['manifests'][$name] = $manifest;
             }
@@ -141,6 +144,7 @@ public function getRecipes(array $operations): array
                 $data['locks'][$name] = $lock;
             }
         }
+
         return $data;
     }
 
@@ -156,17 +160,10 @@ public function get(string $path, array $headers = [], $cache = true): Response
         $url = $this->endpoint.'/'.ltrim($path, '/');
         $cacheKey = $cache ? ltrim($path, '/') : '';
 
-        $rfs = $this->rfs;
-        if ($this->eventDispatcher) {
-            $preFileDownloadEvent = new PreFileDownloadEvent(PluginEvents::PRE_FILE_DOWNLOAD, $rfs, $url);
-            $this->eventDispatcher->dispatch($preFileDownloadEvent->getName(), $preFileDownloadEvent);
-            $rfs = $preFileDownloadEvent->getRemoteFilesystem();
-        }
-
         if ($cacheKey && $contents = $this->cache->read($cacheKey)) {
             $cachedResponse = Response::fromJson(json_decode($contents, true));
             if ($lastModified = $cachedResponse->getHeader('last-modified')) {
-                $response = $this->fetchFileIfLastModified($rfs, $url, $cacheKey, $lastModified, $headers);
+                $response = $this->fetchFileIfLastModified($url, $cacheKey, $lastModified, $headers);
                 if (304 === $response->getStatusCode()) {
                     $response = new Response($cachedResponse->getBody(), $response->getOrigHeaders(), 304);
                 }
@@ -175,18 +172,18 @@ public function get(string $path, array $headers = [], $cache = true): Response
             }
         }
 
-        return $this->fetchFile($rfs, $url, $cacheKey, $headers);
+        return $this->fetchFile($url, $cacheKey, $headers);
     }
 
-    private function fetchFile(RemoteFilesystem $rfs, string $url, string $cacheKey, array $headers): Response
+    private function fetchFile(string $url, string $cacheKey, array $headers): Response
     {
         $options = $this->getOptions($headers);
         $retries = 3;
         while ($retries--) {
             try {
-                $json = $rfs->getContents($this->endpoint, $url, false, $options);
+                $json = $this->rfs->getContents($this->endpoint, $url, false, $options);
 
-                return $this->parseJson($json, $url, $cacheKey, $rfs->getLastHeaders());
+                return $this->parseJson($json, $url, $cacheKey, $this->rfs->getLastHeaders());
             } catch (\Exception $e) {
                 if ($e instanceof TransportException && 404 === $e->getStatusCode()) {
                     throw $e;
@@ -208,19 +205,19 @@ private function fetchFile(RemoteFilesystem $rfs, string $url, string $cacheKey,
         }
     }
 
-    private function fetchFileIfLastModified(RemoteFilesystem $rfs, string $url, string $cacheKey, string $lastModifiedTime, array $headers): Response
+    private function fetchFileIfLastModified(string $url, string $cacheKey, string $lastModifiedTime, array $headers): Response
     {
         $headers[] = 'If-Modified-Since: '.$lastModifiedTime;
         $options = $this->getOptions($headers);
         $retries = 3;
         while ($retries--) {
             try {
-                $json = $rfs->getContents($this->endpoint, $url, false, $options);
-                if (304 === $rfs->findStatusCode($rfs->getLastHeaders())) {
-                    return new Response('', $rfs->getLastHeaders(), 304);
+                $json = $this->rfs->getContents($this->endpoint, $url, false, $options);
+                if (304 === $this->rfs->findStatusCode($this->rfs->getLastHeaders())) {
+                    return new Response('', $this->rfs->getLastHeaders(), 304);
                 }
 
-                return $this->parseJson($json, $url, $cacheKey, $rfs->getLastHeaders());
+                return $this->parseJson($json, $url, $cacheKey, $this->rfs->getLastHeaders());
             } catch (\Exception $e) {
                 if ($e instanceof TransportException && 404 === $e->getStatusCode()) {
                     throw $e;