Commit

README.md

@@ -1 +1,3 @@
-# coroutine
+# Webman coroutine
+
+This is Webman's coroutine library, which includes channel and connection pool.

composer.json

@@ -0,0 +1,14 @@
+{
+  "name": "webman/coroutine",
+  "type": "library",
+  "license": "MIT",
+  "description": "Webman coroutine",
+  "require": {
+    "workerman/workerman": "^5.0.0"
+  },
+  "autoload": {
+    "psr-4": {
+      "Webman\\Coroutine\\": "src"
+    }
+  }
+}

src/Channel.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webman\Coroutine;
+
+use Webman\Coroutine\Channel\ChannelInterface;
+use Webman\Coroutine\Channel\Memory as ChannelMemory;
+use Webman\Coroutine\Channel\Swoole as ChannelSwoole;
+use Webman\Coroutine\Channel\Swow as ChannelSwow;
+use Workerman\Events\Swoole;
+use Workerman\Events\Swow;
+use Workerman\Worker;
+
+class Channel implements ChannelInterface
+{
+    protected ChannelInterface $driver;
+
+    public function __construct(int $capacity)
+    {
+        $this->driver = match (Worker::$eventLoopClass) {
+            Swoole::class => new ChannelSwoole($capacity),
+            Swow::class => new ChannelSwow($capacity),
+            default => new ChannelMemory($capacity),
+        };
+    }
+
+    public function push(mixed $data, float $timeout = -1): bool
+    {
+        return $this->driver->push($data, $timeout);
+    }
+
+    public function pop(float $timeout = -1): mixed
+    {
+        return $this->driver->pop($timeout);
+    }
+
+    public function length(): int
+    {
+        return $this->driver->length();
+    }
+
+    public function getCapacity(): int
+    {
+        return $this->driver->getCapacity();
+    }
+
+    public function close(): void
+    {
+        $this->driver->close();
+    }
+}

src/Channel/ChannelInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webman\Coroutine\Channel;
+
+interface ChannelInterface
+{
+    public function push(mixed $data, float $timeout = -1): bool;
+
+    public function pop(float $timeout = -1): mixed;
+
+    public function length(): int;
+
+    public function getCapacity(): int;
+
+    public function close(): void;
+
+}

src/Channel/Memory.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webman\Coroutine\Channel;
+
+
+class Memory implements ChannelInterface
+{
+
+    protected array $data = [];
+
+    public function __construct(protected int $capacity)
+    {
+    }
+
+    public function push(mixed $data, float $timeout = -1): bool
+    {
+        if ($this->length() >= $this->capacity) {
+            return false;
+        }
+        $this->data[] = $data;
+        return true;
+    }
+
+    public function pop(float $timeout = -1): mixed
+    {
+        if ($this->length() === 0) {
+            return false;
+        }
+        return array_shift($this->data);
+    }
+
+    public function length(): int
+    {
+        return count($this->data);
+    }
+
+    public function getCapacity(): int
+    {
+        return $this->capacity;
+    }
+
+    public function close(): void
+    {
+        $this->data = [];
+    }
+
+}

src/Channel/Swoole.php

@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webman\Coroutine\Channel;
+
+use Swoole\Coroutine\Channel;
+
+class Swoole implements ChannelInterface
+{
+    protected Channel $channel;
+
+    public function __construct(protected int $capacity)
+    {
+        $this->channel = new Channel($capacity);
+    }
+
+    public function push(mixed $data, float $timeout = -1): bool
+    {
+        return $this->channel->push($data, $timeout);
+    }
+
+    public function pop(float $timeout = -1): mixed
+    {
+        return $this->channel->pop($timeout);
+    }
+
+    public function length(): int
+    {
+        return $this->channel->length();
+    }
+
+    public function getCapacity(): int
+    {
+        return $this->channel->capacity;
+    }
+
+    public function close(): void
+    {
+        $this->channel->close();
+    }
+
+}

src/Channel/Swow.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webman\Coroutine\Channel;
+
+use Swow\Channel;
+use Throwable;
+
+class Swow implements ChannelInterface
+{
+    protected Channel $channel;
+
+    public function __construct(protected int $capacity)
+    {
+        $this->channel = new Channel($capacity);
+    }
+
+    public function push(mixed $data, float $timeout = -1): bool
+    {
+        try {
+            $this->channel->push($data, $timeout === -1 ? -1 : (int)$timeout * 1000);
+        } catch (Throwable $e) {
+            return false;
+        }
+        return true;
+    }
+
+    public function pop(float $timeout = -1): mixed
+    {
+        try {
+            return $this->channel->pop($timeout === -1 ? -1 : (int)$timeout * 1000);
+        } catch (Throwable $e) {
+            return false;
+        }
+    }
+
+    public function length(): int
+    {
+        return $this->channel->getLength();
+    }
+
+    public function getCapacity(): int
+    {
+        return $this->channel->getCapacity();
+    }
+
+    public function close(): void
+    {
+        $this->channel->close();
+    }
+
+}