Use function instead of class

amphp · May 4, 2020 · 03dc508 · 03dc508
1 parent 5c2031e
commit 03dc508
Show file tree

Hide file tree

Showing 4 changed files with 58 additions and 73 deletions.
diff --git a/composer.json b/composer.json
@@ -22,7 +22,7 @@
     ],
     "require": {
         "php": ">=7.1",
-        "amphp/amp": "^2.2"
+        "amphp/amp": "dev-discard as 2.5.0"
     },
     "require-dev": {
         "phpunit/phpunit": "^8 || ^7",

diff --git a/examples/queue.php b/examples/queue.php
@@ -4,7 +4,9 @@
 use Amp\Sync\LocalSemaphore;
 use Amp\Sync\Queue;
 use function Amp\delay;
+use function Amp\Iterator\discard;
 use function Amp\Promise\wait;
+use function Amp\Sync\concurrentMap;
 
 require __DIR__ . '/../vendor/autoload.php';
 
@@ -17,7 +19,8 @@
     $emitter->emit($jobId++);
 }
 
-wait(Queue::fromIterator($emitter->iterate())->process(
+wait(discard(concurrentMap(
+    $emitter->iterate(),
     new LocalSemaphore(3),
     function ($job) use ($emitter, &$jobId) {
         print 'starting ' . $job . \PHP_EOL;
@@ -34,8 +37,5 @@ function ($job) use ($emitter, &$jobId) {
         }
 
         print 'finished ' . $job . \PHP_EOL;
-    },
-    function (\Throwable $error, $job) {
-        throw $error;
     }
-));
+)));
diff --git a/src/Queue.php b/src/Queue.php
diff --git a/src/functions.php b/src/functions.php
@@ -2,6 +2,8 @@
 
 namespace Amp\Sync;
 
+use Amp\Iterator;
+use Amp\Producer;
 use Amp\Promise;
 use function Amp\call;
 
@@ -29,3 +31,53 @@ function synchronized(Mutex $mutex, callable $callback, ...$args): Promise
         }
     });
 }
+
+/**
+ * @param Iterator  $iterator
+ * @param Semaphore $semaphore
+ * @param callable  $processor
+ *
+ * @return Iterator
+ */
+function concurrentMap(Iterator $iterator, Semaphore $semaphore, callable $processor): Iterator
+{
+    return new Producer(function () use ($iterator, $semaphore, $processor) {
+        /** @var \Throwable|null $error */
+        $error = null;
+        $pending = [];
+
+        while (yield $iterator->advance()) {
+            if ($error) {
+                throw $error;
+            }
+
+            /** @var Lock $lock */
+            $lock = yield $semaphore->acquire();
+
+            $currentElement = $iterator->getCurrent();
+
+            $promise = call(static function () use ($lock, $currentElement, $processor, &$error) {
+                try {
+                    yield call($processor, $currentElement);
+                } catch (\Throwable $e) {
+                    $error = $e;
+                } finally {
+                    $lock->release();
+                }
+            });
+
+            $promiseId = \spl_object_id($promise);
+
+            $pending[$promiseId] = $promise;
+            $promise->onResolve(static function () use (&$pending, $promiseId) {
+                unset($pending[$promiseId]);
+            });
+        }
+
+        yield Promise\any($pending);
+
+        if ($error) {
+            throw $error;
+        }
+    });
+}