✨ Add scrap

composer.json

@@ -17,7 +17,7 @@
         "ext-iconv": "*",
         "ext-openswoole": "*",
         "amphp/amp": "^3.0",
-        "darkwood/flow": "^1.2",
+        "darkwood/flow": "dev-1.x-dev",
         "doctrine/dbal": "^3",
         "doctrine/doctrine-bundle": "^2.12",
         "doctrine/doctrine-migrations-bundle": "^3.3",

src/Command/ScrapCommand.php

@@ -0,0 +1,82 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Command;
+
+use App\Job\FlowExamples\ScrapUrlJob;
+use App\Job\FlowExamples\ScrapUrlsJob;
+use App\Model\UrlContent;
+use Flow\Driver\FiberDriver;
+use Flow\Flow\Flow;
+use Flow\Ip;
+use Flow\IpStrategy\FlattenIpStrategy;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+use function count;
+use function sprintf;
+
+#[AsCommand(
+    name: 'app:scrap',
+    description: 'This allows scrap pages with flow',
+)]
+class ScrapCommand extends Command
+{
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output);
+
+        $driver = new FiberDriver();
+
+        $flow = Flow::do(static function () use ($io) {
+            yield new ScrapUrlsJob();
+            yield static function (array $urlDatas) use ($io) {
+                $io->writeln(sprintf('ScrapUrlsJob : Finished scrapping %d urls', count($urlDatas)));
+
+                return $urlDatas;
+            };
+            yield [new ScrapUrlJob(), null, new FlattenIpStrategy()];
+            yield static function (UrlContent $urlData) use ($io) {
+               $io->writeln(sprintf('ScrapUrlJob  : Finished scrapping %s', $urlData->url));
+            };
+        }, ['driver' => $driver]);
+
+        $urls = [
+            new UrlContent('https://www.google.fr'),
+            new UrlContent('https://www.apple.com'),
+            new UrlContent('https://www.microsoft.com'),
+            new UrlContent('https://www.amazon.com'),
+            new UrlContent('https://www.facebook.com'),
+            new UrlContent('https://www.netflix.com'),
+            new UrlContent('https://www.spotify.com'),
+            new UrlContent('https://www.wikipedia.org'),
+            new UrlContent('https://www.x.com'),
+            new UrlContent('https://www.instagram.com'),
+            new UrlContent('https://www.linkedin.com'),
+            new UrlContent('https://www.reddit.com'),
+            new UrlContent('https://www.ebay.com'),
+            new UrlContent('https://www.cnn.com'),
+            new UrlContent('https://www.bbc.co.uk'),
+            new UrlContent('https://www.yahoo.com'),
+            new UrlContent('https://www.bing.com'),
+            new UrlContent('https://www.pinterest.com'),
+            new UrlContent('https://www.tumblr.com'),
+            new UrlContent('https://www.paypal.com'),
+            new UrlContent('https://www.dropbox.com'),
+            new UrlContent('https://www.adobe.com'),
+            new UrlContent('https://www.salesforce.com'),
+        ];
+
+        $flow(new Ip($urls));
+
+        $flow->await();
+
+        $io->success('Scraping is done.');
+
+        return Command::SUCCESS;
+    }
+}

src/IpStrategy/FlattenIpStrategy.php

@@ -0,0 +1,71 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\IpStrategy;
+
+use Flow\Event;
+use Flow\Event\PoolEvent;
+use Flow\Event\PullEvent;
+use Flow\Event\PushEvent;
+use Flow\Exception\LogicException;
+use Flow\Ip;
+use Flow\IpPool;
+use Flow\IpStrategyInterface;
+
+/**
+ * @template T
+ *
+ * @implements IpStrategyInterface<T>
+ */
+class FlattenIpStrategy implements IpStrategyInterface
+{
+    /**
+     * @var IpPool<T>
+     */
+    private IpPool $ipPool;
+
+    public function __construct()
+    {
+        $this->ipPool = new IpPool();
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            Event::PUSH => 'push',
+            Event::PULL => 'pull',
+            Event::POOL => 'pool',
+        ];
+    }
+
+    /**
+     * @param PushEvent<T> $event
+     */
+    public function push(PushEvent $event): void
+    {
+        $ip = $event->getIp();
+        if (!is_iterable($ip->data)) {
+            throw new LogicException('Ip data must be iterable');
+        }
+        foreach ($ip->data as $data) {
+            $this->ipPool->addIp(new Ip($data));
+        }
+    }
+
+    /**
+     * @param PullEvent<T> $event
+     */
+    public function pull(PullEvent $event): void
+    {
+        $ip = $this->ipPool->shiftIp();
+        if ($ip !== null) {
+            $event->addIp($ip);
+        }
+    }
+
+    public function pool(PoolEvent $event): void
+    {
+        $event->addIps($this->ipPool->getIps());
+    }
+}

src/Job/ScrapUrlJob.php

@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Job\FlowExamples;
+
+use App\Model\UrlContent;
+use CurlMultiHandle;
+use Fiber;
+use Flow\JobInterface;
+
+/**
+ * @implements JobInterface<UrlContent, UrlContent>
+ */
+class ScrapUrlJob implements JobInterface
+{
+    private CurlMultiHandle $mh;
+
+    public function __construct()
+    {
+        // Initialize a cURL multi handle
+        $this->mh = curl_multi_init();
+    }
+
+    public function __destruct()
+    {
+        curl_multi_close($this->mh);
+    }
+
+    public function __invoke($urlContent): mixed
+    {
+        $ch = curl_init();
+        curl_setopt($ch, CURLOPT_URL, $urlContent->url);
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+        curl_multi_add_handle($this->mh, $ch);
+
+        do {
+            $status = curl_multi_exec($this->mh, $active);
+            curl_multi_exec($this->mh, $active);
+
+            Fiber::suspend();
+
+            $info = curl_multi_info_read($this->mh);
+        } while (
+            $active && $status === CURLM_OK // check curl_multi is active
+            && !($info !== false && $info['handle'] === $ch && $info['result'] === CURLE_OK) // check $ch is done
+        );
+
+        $content = curl_multi_getcontent($ch);
+        curl_multi_remove_handle($this->mh, $ch);
+        curl_close($ch);
+
+        return new UrlContent($urlContent->url, $content);
+    }
+}