✨ Add event system for processing IpStrategy

CHANGELOG.md

@@ -8,6 +8,8 @@
     - array : constructor arguments for Flow instanciation
     - array (view as shape) : configuration for Flow instanciation
     - FlowInterface : the FlowInterface instance itself
+    - array : map of all possible above choices
+- Add event system for processing IpStrategy
 
 ## v1.1.4
 

composer.json

@@ -21,7 +21,8 @@
         "php"
     ],
     "require": {
-        "php": ">=8.2"
+        "php": ">=8.2",
+        "symfony/event-dispatcher": "^6.3"
     },
     "require-dev": {
         "amphp/amp": "^3.0",

src/Event/PopEvent.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Event;
+
+use Flow\Ip;
+use Symfony\Contracts\EventDispatcher\Event;
+
+/**
+ * @template T
+ */
+final class PopEvent extends Event
+{
+    /**
+     * @param Ip<T> $ip
+     */
+    private Ip $ip;
+
+    /**
+     * @param Ip<T> $ip
+     */
+    public function __construct(Ip $ip)
+    {
+        $this->ip = $ip;
+    }
+
+    /**
+     * @return Ip<T>
+     */
+    public function getIp(): Ip
+    {
+        return $this->ip;
+    }
+}

src/Event/PullEvent.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Event;
+
+use Flow\Ip;
+use Symfony\Contracts\EventDispatcher\Event;
+
+/**
+ * @template T
+ */
+final class PullEvent extends Event
+{
+    /**
+     * @param null|Ip<T> $ip
+     */
+    private ?Ip $ip = null;
+
+    /**
+     * @return null|Ip<T>
+     */
+    public function getIp(): ?Ip
+    {
+        return $this->ip;
+    }
+
+    /**
+     * @return null|Ip<T>
+     */
+    public function setIp(?Ip $ip)
+    {
+        $this->ip = $ip;
+    }
+}

src/Event/PushEvent.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Event;
+
+use Flow\Ip;
+use Symfony\Contracts\EventDispatcher\Event;
+
+/**
+ * @template T
+ */
+final class PushEvent extends Event
+{
+    /**
+     * @param Ip<T> $ip
+     */
+    private Ip $ip;
+
+    /**
+     * @param Ip<T> $ip
+     */
+    public function __construct(Ip $ip)
+    {
+        $this->ip = $ip;
+    }
+
+    /**
+     * @return Ip<T>
+     */
+    public function getIp(): Ip
+    {
+        return $this->ip;
+    }
+}

src/Flow/Flow.php

@@ -7,14 +7,20 @@
 use Closure;
 use Flow\Driver\FiberDriver;
 use Flow\DriverInterface;
+use Flow\Event\PopEvent;
+use Flow\Event\PullEvent;
+use Flow\Event\PushEvent;
 use Flow\Exception\LogicException;
 use Flow\ExceptionInterface;
 use Flow\FlowInterface;
 use Flow\Ip;
 use Flow\IpStrategy\LinearIpStrategy;
+use Flow\IpStrategyEvent;
 use Flow\IpStrategyInterface;
 use Generator;
 use SplObjectStorage;
+use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 use function array_key_exists;
 use function count;
@@ -48,6 +54,8 @@ class Flow implements FlowInterface
      */
     private DriverInterface $driver;
 
+    private EventDispatcherInterface $dispatcher;
+
     /**
      * @var SplObjectStorage<Ip<T1>, null|Closure(Ip<T1>): void>
      */
@@ -68,19 +76,22 @@ public function __construct(
         Closure|array $jobs,
         Closure|array $errorJobs = null,
         IpStrategyInterface $ipStrategy = null,
-        DriverInterface $driver = null
+        DriverInterface $driver = null,
+        EventDispatcherInterface $dispatcher = null,
     ) {
         $this->jobs = is_array($jobs) ? $jobs : [$jobs];
         $this->errorJobs = $errorJobs ? (is_array($errorJobs) ? $errorJobs : [$errorJobs]) : [];
         $this->ipStrategy = $ipStrategy ?? new LinearIpStrategy();
         $this->driver = $driver ?? new FiberDriver();
+        $this->dispatcher = $dispatcher ?? new EventDispatcher();
+        $this->dispatcher->addSubscriber($this->ipStrategy);
         $this->callbacks = new SplObjectStorage();
     }
 
     public function __invoke(Ip $ip, Closure $callback = null): void
     {
         $this->callbacks->offsetSet($ip, $callback);
-        $this->ipStrategy->push($ip);
+        $this->dispatcher->dispatch(new PushEvent($ip), IpStrategyEvent::PUSH);
         $this->nextIpJob();
     }
 
@@ -128,7 +139,7 @@ public function fn(array|Closure|FlowInterface $flow): FlowInterface
 
     private function nextIpJob(): void
     {
-        $ip = $this->ipStrategy->pop();
+        $ip = $this->dispatcher->dispatch(new PullEvent(), IpStrategyEvent::PULL)->getIp();
         if (!$ip) {
             return;
         }
@@ -142,7 +153,7 @@ private function nextIpJob(): void
                 $count--;
                 if ($count === 0 || $value instanceof ExceptionInterface) {
                     $count = 0;
-                    $this->ipStrategy->done($ip);
+                    $this->dispatcher->dispatch(new PopEvent($ip), IpStrategyEvent::POP);
                     $this->nextIpJob();
 
                     if ($value instanceof ExceptionInterface) {

src/IpStrategy/LinearIpStrategy.php

@@ -4,13 +4,14 @@
 
 namespace Flow\IpStrategy;
 
+use Flow\Event\PullEvent;
+use Flow\Event\PushEvent;
 use Flow\Ip;
+use Flow\IpStrategyEvent;
 use Flow\IpStrategyInterface;
 
 /**
  * @template T
- *
- * @implements IpStrategyInterface<T>
  */
 class LinearIpStrategy implements IpStrategyInterface
 {
@@ -19,26 +20,27 @@ class LinearIpStrategy implements IpStrategyInterface
      */
     private array $ips = [];
 
-    /**
-     * @param Ip<T> $ip
-     */
-    public function push(Ip $ip): void
+    public static function getSubscribedEvents(): array
     {
-        $this->ips[] = $ip;
+        return [
+            IpStrategyEvent::PUSH => 'push',
+            IpStrategyEvent::PULL => 'pull',
+        ];
     }
 
     /**
-     * @return null|Ip<T>
+     * @param PushEvent<T> $event
      */
-    public function pop(): ?Ip
+    public function push(PushEvent $event): void
     {
-        return array_shift($this->ips);
+        $this->ips[] = $event->getIp();
     }
 
     /**
-     * @param Ip<T> $ip
+     * @param PullEvent<T> $event
      */
-    public function done(Ip $ip): void
+    public function pull(PullEvent $event): void
     {
+        $event->setIp(array_shift($this->ips));
     }
 }

src/IpStrategy/MaxIpStrategy.php

@@ -4,62 +4,70 @@
 
 namespace Flow\IpStrategy;
 
-use Flow\Ip;
+use Flow\Event\PopEvent;
+use Flow\Event\PullEvent;
+use Flow\Event\PushEvent;
+use Flow\IpStrategyEvent;
 use Flow\IpStrategyInterface;
+use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 /**
  * @template T
- *
- * @implements IpStrategyInterface<T>
  */
 class MaxIpStrategy implements IpStrategyInterface
 {
-    /**
-     * @var IpStrategyInterface<T>
-     */
     private IpStrategyInterface $ipStrategy;
 
+    private EventDispatcherInterface $dispatcher;
+
     private int $processing = 0;
 
-    /**
-     * @param null|IpStrategyInterface<T> $ipStrategy
-     */
-    public function __construct(private int $max = 1, IpStrategyInterface $ipStrategy = null)
-    {
+    public function __construct(
+        private int $max = 1,
+        ?IpStrategyInterface $ipStrategy = null,
+        EventDispatcherInterface $dispatcher = null,
+    ) {
         $this->ipStrategy = $ipStrategy ?? new LinearIpStrategy();
+        $this->dispatcher = $dispatcher ?? new EventDispatcher();
+        $this->dispatcher->addSubscriber($this->ipStrategy);
     }
 
-    /**
-     * @param Ip<T> $ip
-     */
-    public function push(Ip $ip): void
+    public static function getSubscribedEvents()
+    {
+        return [
+            IpStrategyEvent::PUSH => 'push',
+            IpStrategyEvent::PULL => 'pull',
+            IpStrategyEvent::POP => 'pop',
+        ];
+    }
+
+    public function push(PushEvent $event): void
     {
-        $this->ipStrategy->push($ip);
+        $this->dispatcher->dispatch($event, IpStrategyEvent::PUSH);
     }
 
     /**
-     * @return null|Ip<T>
+     * @return PullEvent<T>
      */
-    public function pop(): ?Ip
+    public function pull(PullEvent $event): void
     {
         if ($this->processing < $this->max) {
-            $ip = $this->ipStrategy->pop();
+            $ip = $this->dispatcher->dispatch($event, IpStrategyEvent::PULL)->getIp();
             if ($ip) {
                 $this->processing++;
             }
 
-            return $ip;
+            $event->setIp($ip);
         }
-
-        return null;
     }
 
     /**
-     * @param Ip<T> $ip
+     * @param PopEvent<T> $event
      */
-    public function done(Ip $ip): void
+    public function pop(PopEvent $event): void
     {
-        $this->ipStrategy->done($ip);
+        $this->dispatcher->dispatch($event, IpStrategyEvent::POP);
         $this->processing--;
     }
 }