Fixed redis connection has already been bound to another coroutine. (#2582)

* Added test cases.

* Update

* Update CHANGELOG-2.0.md

CHANGELOG-2.0.md

@@ -10,6 +10,7 @@
 - [#2561](https://github.com/hyperf/hyperf/pull/2561) Optimized error message when close amqp connection failed.
 - [#2565](https://github.com/hyperf/hyperf/pull/2565) Fixed proxy class generate keyword `parent::class` but the class scope has on parent.
 - [#2578](https://github.com/hyperf/hyperf/pull/2578) Fixed event `AfterProcessHandle` won't be dispatched when throw exception in process.
+- [#2582](https://github.com/hyperf/hyperf/pull/2582) Fixed redis connection has already been bound to another coroutine.
 
 # v2.0.12 - 2020-09-21
 

src/redis/src/Redis.php

@@ -57,6 +57,7 @@ class Redis
                     // Should storage the connection to coroutine context, then use defer() to release the connection.
                     Context::set($this->getContextKey(), $connection);
                     defer(function () use ($connection) {
+                        Context::set($this->getContextKey(), null);
                         $connection->release();
                     });
                 } else {

src/redis/tests/RedisProxyTest.php

@@ -113,7 +113,7 @@ class RedisProxyTest extends TestCase
         $container->shouldReceive('get')->once()->with(ConfigInterface::class)->andReturn(new Config([
             'redis' => [
                 'default' => [
-                    'host' => 'localhost',
+                    'host' => '127.0.0.1',
                     'auth' => null,
                     'port' => 6379,
                     'db' => 0,

src/redis/tests/RedisTest.php

@@ -22,6 +22,7 @@ use Hyperf\Redis\Pool\RedisPool;
 use Hyperf\Redis\Redis;
 use Hyperf\Utils\ApplicationContext;
 use Hyperf\Utils\Context;
+use Hyperf\Utils\Coroutine;
 use HyperfTest\Redis\Stub\RedisPoolFailedStub;
 use HyperfTest\Redis\Stub\RedisPoolStub;
 use Mockery;
@@ -73,6 +74,52 @@ class RedisTest extends TestCase
         $this->assertSame('db:0 name:get argument:xxxx', $res[0]);
     }
 
+    public function testHasAlreadyBeenBoundToAnotherCoroutine()
+    {
+        $chan = new \Swoole\Coroutine\Channel(1);
+        $redis = $this->getRedis();
+        $ref = new \ReflectionClass($redis);
+        $method = $ref->getMethod('getConnection');
+        $method->setAccessible(true);
+
+        go(function () use ($chan, $redis, $method) {
+            $id = null;
+            defer(function () use ($redis, $chan, &$id, $method) {
+                $chan->push(true);
+                Coroutine::sleep(0.01);
+                $hasContextConnection = Context::has('redis.connection.default');
+                $this->assertFalse($hasContextConnection);
+                $connection = $method->invoke($redis, [$hasContextConnection]);
+                $this->assertNotEquals($connection->id, $id);
+                $redis->getConnection();
+                $chan->push(true);
+            });
+
+            $redis->keys('*');
+
+            $hasContextConnection = Context::has('redis.connection.default');
+            $this->assertFalse($hasContextConnection);
+
+            $redis->multi();
+            $redis->set('id', uniqid());
+            $redis->exec();
+
+            $hasContextConnection = Context::has('redis.connection.default');
+            $this->assertTrue($hasContextConnection);
+
+            $connection = $method->invoke($redis, [$hasContextConnection]);
+            $id = $connection->id;
+        });
+
+        $chan->pop();
+        $factory = $ref->getProperty('factory');
+        $factory->setAccessible(true);
+        $factory = $factory->getValue($redis);
+        $pool = $factory->getPool('default');
+        $pool->flushAll();
+        $chan->pop();
+    }
+
     public function testRedisReuseAfterThrowable()
     {
         $container = $this->getContainer();
@@ -108,7 +155,7 @@ class RedisTest extends TestCase
         $container->shouldReceive('get')->once()->with(ConfigInterface::class)->andReturn(new Config([
             'redis' => [
                 'default' => [
-                    'host' => 'localhost',
+                    'host' => '127.0.0.1',
                     'auth' => null,
                     'port' => 6379,
                     'db' => 0,

src/redis/tests/Stub/RedisConnectionStub.php

@@ -11,7 +11,9 @@ declare(strict_types=1);
  */
 namespace HyperfTest\Redis\Stub;
 
+use Hyperf\Pool\Pool;
 use Hyperf\Redis\RedisConnection;
+use Psr\Container\ContainerInterface;
 
 class RedisConnectionStub extends RedisConnection
 {
@@ -25,6 +27,14 @@ class RedisConnectionStub extends RedisConnection
 
     public $timeout;
 
+    public $id;
+
+    public function __construct(ContainerInterface $container, Pool $pool, array $config)
+    {
+        parent::__construct($container, $pool, $config);
+        $this->id = uniqid();
+    }
+
     public function __call($name, $arguments)
     {
         return sprintf('db:%d name:%s argument:%s', $this->db, $name, implode(',', $arguments));

src/redis/tests/Stub/RedisPoolStub.php

@@ -12,10 +12,26 @@ declare(strict_types=1);
 namespace HyperfTest\Redis\Stub;
 
 use Hyperf\Contract\ConnectionInterface;
+use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Redis\Pool\RedisPool;
 
 class RedisPoolStub extends RedisPool
 {
+    public function flushAll()
+    {
+        while ($conn = $this->channel->pop(0.001)) {
+            try {
+                $conn->close();
+            } catch (\Throwable $exception) {
+                if ($this->container->has(StdoutLoggerInterface::class) && $logger = $this->container->get(StdoutLoggerInterface::class)) {
+                    $logger->error((string) $exception);
+                }
+            } finally {
+                --$this->currentConnections;
+            }
+        }
+    }
+
     protected function createConnection(): ConnectionInterface
     {
         return new RedisConnectionStub($this->container, $this, $this->config);