rate limit

composer.json

@@ -35,6 +35,7 @@
         "openzipkin/zipkin": "^1.3.2",
         "grpc/grpc": "^1.15",
         "elasticsearch/elasticsearch": "^6.1",
+        "bandwidth-throttle/token-bucket": "^2.0",
         "monolog/monolog": "^1.24"
     },
     "require-dev": {
@@ -71,6 +72,7 @@
         "hyperf/pool": "self.version",
         "hyperf/process": "self.version",
         "hyperf/queue": "self.version",
+        "hyperf/rate-limit": "self.version",
         "hyperf/redis": "self.version",
         "hyperf/tracer": "self.version",
         "hyperf/utils": "self.version"
@@ -109,6 +111,7 @@
             "Hyperf\\Pool\\": "src/pool/src/",
             "Hyperf\\Process\\": "src/process/src/",
             "Hyperf\\Queue\\": "src/queue/src/",
+            "Hyperf\\RateLimit\\": "src/rate-limit/src/",
             "Hyperf\\Redis\\": "src/redis/src/",
             "Hyperf\\Tracer\\": "src/tracer/src/",
             "Hyperf\\Utils\\": "src/utils/src/"
@@ -152,6 +155,7 @@
                 "Hyperf\\Pool\\ConfigProvider",
                 "Hyperf\\Process\\ConfigProvider",
                 "Hyperf\\Queue\\ConfigProvider",
+                "Hyperf\\RateLimit\\ConfigProvider",
                 "Hyperf\\Redis\\ConfigProvider",
                 "Hyperf\\Tracer\\ConfigProvider",
                 "Hyperf\\Utils\\ConfigProvider"

src/rate-limit/composer.json

@@ -0,0 +1,55 @@
+{
+    "name": "hyperf/rate-limit",
+    "description": "A rate limiter implemented for Hyperf or other coroutine framework",
+    "license": "Apache-2.0",
+    "keywords": [
+        "php",
+        "hyperf",
+        "rate-limiter",
+        "token-bucket"
+    ],
+    "support": {
+    },
+    "require": {
+        "php": ">=7.2",
+        "hyperf/di": "dev-master",
+        "hyperf/event": "dev-master",
+        "hyperf/redis": "dev-master",
+        "bandwidth-throttle/token-bucket": "^2.0"
+    },
+    "require-dev": {
+        "malukenho/docheader": "^0.1.6",
+        "mockery/mockery": "^1.0",
+        "phpunit/phpunit": "^7.0.0",
+        "friendsofphp/php-cs-fixer": "^2.9"
+    },
+    "suggest": {
+    },
+    "autoload": {
+        "files": [
+        ],
+        "psr-4": {
+            "Hyperf\\RateLimit\\": "src/"
+        }
+    },
+    "autoload-dev": {
+        "psr-4": {
+        }
+    },
+    "config": {
+        "sort-packages": true
+    },
+    "extra": {
+        "branch-alias": {
+        },
+        "hyperf": {
+            "config": "Hyperf\\RateLimit\\ConfigProvider"
+        }
+    },
+    "bin": [
+    ],
+    "scripts": {
+        "cs-fix": "php-cs-fixer fix $1",
+        "test": "phpunit --colors=always"
+    }
+}

src/rate-limit/src/Annotation/RateLimit.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.org
+ * @document https://wiki.hyperf.org
+ * @contact  group@hyperf.org
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\RateLimit\Annotation;
+
+use Doctrine\Common\Annotations\Annotation\Target;
+use Hyperf\Di\Annotation\AbstractAnnotation;
+
+/**
+ * @Annotation
+ * @Target({"CLASS", "METHOD"})
+ */
+class RateLimit extends AbstractAnnotation
+{
+    /**
+     * @var int
+     */
+    public $limit;
+
+    /**
+     * @var int
+     */
+    public $demand;
+
+    /**
+     * @var int
+     */
+    public $capacity;
+
+    /**
+     * @var array
+     */
+    public $callback;
+
+    /**
+     * @var callable|string
+     */
+    public $bucketsKey;
+}

src/rate-limit/src/Aspect/RateLimitAnnotationAspect.php

@@ -0,0 +1,119 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.org
+ * @document https://wiki.hyperf.org
+ * @contact  group@hyperf.org
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\RateLimit\Aspect;
+
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\Di\Annotation\Aspect;
+use Hyperf\Di\Aop\ArroundInterface;
+use Hyperf\Di\Aop\ProceedingJoinPoint;
+use Hyperf\HttpServer\Contract\RequestInterface;
+use Hyperf\RateLimit\Annotation\RateLimit;
+use Hyperf\RateLimit\Exception\RateLimiterException;
+use Hyperf\RateLimit\Handler\RateLimitHandler;
+
+/**
+ * @Aspect
+ */
+class RateLimitAnnotationAspect implements ArroundInterface
+{
+    public $classes = [];
+
+    public $annotations = [
+        RateLimit::class,
+    ];
+
+    /**
+     * @var array
+     */
+    private $annotationProperty;
+
+    /**
+     * @var ConfigInterface
+     */
+    private $config;
+
+    /**
+     * @var RequestInterface
+     */
+    private $request;
+
+    /**
+     * @var RateLimitHandler
+     */
+    private $rateLimitHandler;
+
+    public function __construct(ConfigInterface $config, RequestInterface $request, RateLimitHandler $rateLimitHandler)
+    {
+        $this->annotationProperty = get_object_vars(new RateLimit());
+        $this->config = $config;
+        $this->request = $request;
+        $this->rateLimitHandler = $rateLimitHandler;
+    }
+
+    /**
+     * @param ProceedingJoinPoint $proceedingJoinPoint
+     * @throws \bandwidthThrottle\tokenBucket\storage\StorageException
+     * @return mixed
+     */
+    public function process(ProceedingJoinPoint $proceedingJoinPoint)
+    {
+        $annotation = $this->getWeightingAnnotation($this->getAnnotations($proceedingJoinPoint));
+
+        $bucketsKey = $annotation->bucketsKey;
+        if (is_callable($bucketsKey)) {
+            $bucketsKey = $bucketsKey($proceedingJoinPoint);
+        }
+        if (! $bucketsKey) {
+            $bucketsKey = trim(str_replace('/', ':', $this->request->getUri()->getPath()), ':');
+        }
+
+        $bucket = $this->rateLimitHandler->getBucket($bucketsKey);
+        if (! $bucket) {
+            $bucket = $this->rateLimitHandler->build($bucketsKey, $annotation->limit, $annotation->capacity);
+        }
+
+        if ($bucket->consume($annotation->demand, $seconds)) {
+            return $proceedingJoinPoint->process();
+        }
+
+        if (! $annotation->callback || ! is_callable($annotation->callback)) {
+            throw new RateLimiterException('Request rate limit');
+        }
+        return call_user_func($annotation->callback, $seconds, $proceedingJoinPoint);
+    }
+
+    /**
+     * @param RateLimit[] $annotations
+     * @return RateLimit
+     */
+    public function getWeightingAnnotation(array $annotations)
+    {
+        $property = array_merge($this->annotationProperty, $this->config->get('rate-limit', []));
+        foreach ($annotations as $annotation) {
+            if (! $annotation) {
+                continue;
+            }
+            $property = array_merge($property, array_filter(get_object_vars($annotation)));
+        }
+        return new RateLimit($property);
+    }
+
+    public function getAnnotations(ProceedingJoinPoint $proceedingJoinPoint)
+    {
+        $metadata = $proceedingJoinPoint->getAnnotationMetadata();
+        return [
+            $metadata->class[RateLimit::class] ?? null,
+            $metadata->method[RateLimit::class] ?? null,
+        ];
+    }
+}

src/rate-limit/src/Exception/RateLimiterException.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.org
+ * @document https://wiki.hyperf.org
+ * @contact  group@hyperf.org
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\RateLimit\Exception;
+
+use RuntimeException;
+
+class RateLimiterException extends RuntimeException
+{
+}

src/rate-limit/src/Handler/RateLimitHandler.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.org
+ * @document https://wiki.hyperf.org
+ * @contact  group@hyperf.org
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\RateLimit\Handler;
+
+use bandwidthThrottle\tokenBucket\Rate;
+use bandwidthThrottle\tokenBucket\TokenBucket;
+use Hyperf\RateLimit\Storage\CoRedisStorage;
+use Psr\Container\ContainerInterface;
+use Psr\SimpleCache\CacheInterface;
+
+class RateLimitHandler
+{
+    const RATE_LIMIT_BUCKETS = 'rateLimit:buckets';
+
+    /**
+     * @var TokenBucket[]
+     */
+    private $buckets;
+
+    /**
+     * @var \Redis
+     */
+    private $redis;
+
+    public function __construct(ContainerInterface $container)
+    {
+        $this->redis = $container->get(\Redis::class);
+    }
+
+    /**
+     * @param string $key
+     * @param int $limit
+     * @param int $capacity
+     * @throws \bandwidthThrottle\tokenBucket\storage\StorageException
+     * @return TokenBucket
+     */
+    public function build(string $key, int $limit, int $capacity)
+    {
+        $storage = make(CoRedisStorage::class, ['key' => $key, 'redis' => $this->redis]);
+        $rate = make(Rate::class, ['tokens' => $limit, 'unit' => Rate::SECOND]);
+        $bucket = make(TokenBucket::class, ['capacity' => $capacity, 'rate' => $rate, 'storage' => $storage]);
+        $bucket->bootstrap($capacity);
+        $this->setBucket($key, $bucket);
+        return $bucket;
+    }
+
+    /**
+     * @param string $key
+     * @return null|TokenBucket
+     */
+    public function getBucket(string $key)
+    {
+        return $this->buckets[$key] ?? null;
+//        return unserialize($this->redis->hGet(self::RATE_LIMIT_BUCKETS, $key) ?: '');
+    }
+
+    /**
+     * @param string $key
+     * @param TokenBucket $bucket
+     */
+    public function setBucket(string $key, TokenBucket $bucket)
+    {
+        $this->buckets[$key] = $bucket;
+//        $this->redis->hSet(self::RATE_LIMIT_BUCKETS, $key, serialize($bucket));
+    }
+}

src/rate-limit/src/Storage/CoRedisStorage.php

@@ -0,0 +1,123 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.org
+ * @document https://wiki.hyperf.org
+ * @contact  group@hyperf.org
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\RateLimit\Storage;
+
+use bandwidthThrottle\tokenBucket\storage\scope\GlobalScope;
+use bandwidthThrottle\tokenBucket\storage\Storage;
+use bandwidthThrottle\tokenBucket\storage\StorageException;
+use bandwidthThrottle\tokenBucket\util\DoublePacker;
+use malkusch\lock\mutex\Mutex;
+use malkusch\lock\mutex\PHPRedisMutex;
+use Psr\SimpleCache\InvalidArgumentException;
+use Redis;
+
+class CoRedisStorage implements Storage, GlobalScope
+{
+    const KEY_PREFIX = 'rateLimiter:storage:';
+
+    /**
+     * @var Mutex
+     */
+    private $mutex;
+
+    /**
+     * @var Redis
+     */
+    private $redis;
+
+    /**
+     * @var string the key
+     */
+    private $key;
+
+    public function __construct($key, $redis)
+    {
+        $key = self::KEY_PREFIX . $key;
+        $this->key = $key;
+        $this->redis = $redis;
+        $this->mutex = make(PHPRedisMutex::class, [
+            'redisAPIs' => [$redis],
+            'name' => $key
+        ]);
+    }
+
+    public function bootstrap($microtime)
+    {
+        $this->setMicrotime($microtime);
+    }
+
+    public function isBootstrapped()
+    {
+        try {
+            return $this->redis->exists($this->key);
+        } catch (InvalidArgumentException $e) {
+            throw new StorageException('Failed to check for key existence', 0, $e);
+        }
+    }
+
+    public function remove()
+    {
+        try {
+            if (! $this->redis->delete($this->key)) {
+                throw new StorageException('Failed to delete key');
+            }
+        } catch (InvalidArgumentException $e) {
+            throw new StorageException('Failed to delete key', 0, $e);
+        }
+    }
+
+    /**
+     * @SuppressWarnings(PHPMD)
+     * @param float $microtime
+     * @throws StorageException
+     */
+    public function setMicrotime($microtime)
+    {
+        try {
+            $data = DoublePacker::pack($microtime);
+
+            if (! $this->redis->set($this->key, $data)) {
+                throw new StorageException('Failed to store microtime');
+            }
+        } catch (InvalidArgumentException $e) {
+            throw new StorageException('Failed to store microtime', 0, $e);
+        }
+    }
+
+    /**
+     * @SuppressWarnings(PHPMD)
+     * @throws StorageException
+     * @return float
+     */
+    public function getMicrotime()
+    {
+        try {
+            $data = $this->redis->get($this->key);
+            if ($data === false) {
+                throw new StorageException('Failed to get microtime');
+            }
+            return DoublePacker::unpack($data);
+        } catch (InvalidArgumentException $e) {
+            throw new StorageException('Failed to get microtime', 0, $e);
+        }
+    }
+
+    public function getMutex()
+    {
+        return $this->mutex;
+    }
+
+    public function letMicrotimeUnchanged()
+    {
+    }
+}