[transport] Add redis backed transport.

Author: makasim

.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 1
+
+language: php
+
+php:
+    - '5.6'
+    - '7.0'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install --prefer-source
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional

LICENSE

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+Copyright (c) 2017 Forma-Pro
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

PRedis.php

@@ -0,0 +1,66 @@
+<?php
+namespace Enqueue\Redis;
+
+use Predis\ClientInterface;
+use Predis\Response\ServerException as PRedisServerException;
+
+class PRedis implements Redis
+{
+    /**
+     * @var ClientInterface
+     */
+    private $redis;
+
+    /**
+     * @param ClientInterface $redis
+     */
+    public function __construct(ClientInterface $redis)
+    {
+        $this->redis = $redis;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function lpush($key, $value)
+    {
+        try {
+            $this->redis->lpush($key, [$value]);
+        } catch (PRedisServerException $e) {
+            throw new ServerException('lpush command has failed', null, $e);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function brpop($key, $timeout)
+    {
+        try {
+            return $this->brpop($key, (int) $timeout / 1000);
+        } catch (PRedisServerException $e) {
+            throw new ServerException('brpop command has failed', null, $e);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function rpop($key)
+    {   try {
+           return $this->rpop($key);
+        } catch (PRedisServerException $e) {
+            throw new ServerException('rpop command has failed', null, $e);
+        }
+    }
+
+    public function connect()
+    {
+        $this->redis->connect();
+    }
+
+    public function disconnect()
+    {
+        $this->redis->disconnect();
+    }
+}

PhpRedis.php

@@ -0,0 +1,93 @@
+<?php
+namespace Enqueue\Redis;
+
+class PhpRedis implements Redis
+{
+    /**
+     * @var \Redis
+     */
+    private $redis;
+
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * @param \Redis $redis
+     * @param array $config
+     */
+    public function __construct(\Redis $redis, array $config)
+    {
+        $this->redis = $redis;
+
+        $this->config = array_replace([
+            'host' => null,
+            'port' => null,
+            'timeout' => null,
+            'reserved' => null,
+            'retry_interval' => null,
+            'persisted' => false,
+        ], $config);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function lpush($key, $value)
+    {
+        if (false == $this->redis->lPush($key, $value)) {
+            throw new ServerException($this->redis->getLastError());
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function brpop($key, $timeout)
+    {
+        if ($result = $this->redis->brPop([$key], $timeout)) {
+            return $result[1];
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function rpop($key)
+    {
+        return $this->redis->rPop($key);
+    }
+
+    public function connect()
+    {
+        if (false == $this->redis) {
+            $this->redis = new \Redis();
+
+            if ($this->config['persisted']) {
+                $this->redis->pconnect(
+                    $this->config['host'],
+                    $this->config['port'],
+                    $this->config['timeout']
+                );
+            } else {
+                $this->redis->connect(
+                    $this->config['host'],
+                    $this->config['port'],
+                    $this->config['timeout'],
+                    $this->config['reserved'],
+                    $this->config['retry_interval']
+                );
+            }
+        }
+
+        return $this->redis;
+    }
+
+    public function disconnect()
+    {
+        if ($this->redis) {
+            $this->redis->close();
+        }
+    }
+}

README.md

@@ -0,0 +1,18 @@
+# Redis Transport
+
+[![Gitter](https://badges.gitter.im/php-enqueue/Lobby.svg)](https://gitter.im/php-enqueue/Lobby)
+[![Build Status](https://travis-ci.org/php-enqueue/redis-ext.png?branch=master)](https://travis-ci.org/php-enqueue/redis-ext)
+[![Total Downloads](https://poser.pugx.org/enqueue/redis-ext/d/total.png)](https://packagist.org/packages/enqueue/redis-ext)
+[![Latest Stable Version](https://poser.pugx.org/enqueue/redis-ext/version.png)](https://packagist.org/packages/enqueue/redis-ext)
+ 
+This is an implementation of PSR specification. It allows you to send and consume message with Redis store as a broker.  
+
+## Resources
+
+* [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
+* [Questions](https://gitter.im/php-enqueue/Lobby)
+* [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
+
+## License
+
+It is released under the [MIT License](LICENSE).

Redis.php

@@ -0,0 +1,32 @@
+<?php
+namespace Enqueue\Redis;
+
+interface Redis
+{
+    /**
+     * @param string $key
+     * @param string $value
+     *
+     * @return int length of the list
+     */
+    public function lpush($key, $value);
+
+    /**
+     * @param string $key
+     * @param int $timeout in seconds
+     *
+     * @return string|null
+     */
+    public function brpop($key, $timeout);
+
+    /**
+     * @param string $key
+     *
+     * @return string|null
+     */
+    public function rpop($key);
+
+    public function connect();
+
+    public function disconnect();
+}

RedisConnectionFactory.php

@@ -0,0 +1,91 @@
+<?php
+
+namespace Enqueue\Redis;
+
+use Enqueue\Psr\PsrConnectionFactory;
+use Predis\Client;
+
+class RedisConnectionFactory implements PsrConnectionFactory
+{
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * @var \Redis
+     */
+    private $redis;
+
+    /**
+     * $config = [
+     *  'host' => can be a host, or the path to a unix domain socket
+     *  'port' => optional
+     *  'timeout' => value in seconds (optional, default is 0.0 meaning unlimited)
+     *  'reserved' => should be null if $retry_interval is specified
+     *  'retry_interval' => retry interval in milliseconds.
+     *  'vendor' => 'The library used internally to interact with Redis server
+     *  'persisted' => bool, Whether it use single persisted connection or open a new one for every context
+     *  'lazy' => the connection will be performed as later as possible, if the option set to true
+     * ].
+     *
+     * @param $config
+     */
+    public function __construct(array $config)
+    {
+        $this->config = array_replace([
+            'host' => null,
+            'port' => null,
+            'timeout' => null,
+            'reserved' => null,
+            'retry_interval' => null,
+            'vendor' => 'phpredis',
+            'persisted' => false,
+            'lazy' => true,
+        ], $config);
+
+        $supportedVendors = ['predis', 'phpredis'];
+        if (false == in_array($this->config['vendor'], $supportedVendors)) {
+            throw new \LogicException(sprintf(
+                'Unsupported redis vendor given. It must be either "%s". Got "%s"',
+                implode('", "', $supportedVendors),
+                $this->config['vendor']
+            ));
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RedisContext
+     */
+    public function createContext()
+    {
+        if ($this->config['lazy']) {
+            return new RedisContext(function () {
+                $redis = $this->createRedis();
+                $redis->connect();
+
+                return $redis;
+            });
+        }
+
+        return new RedisContext($this->createRedis());
+    }
+
+    /**
+     * @return Redis
+     */
+    private function createRedis()
+    {
+        if ('phpredis' == $this->config['vendor'] && false == $this->redis) {
+            $this->redis = new PhpRedis(new \Redis(), $this->config);
+        }
+
+        if ('predis' == $this->config['vendor'] && false == $this->redis) {
+            $this->redis = new PRedis(new Client($this->config, ['exceptions' => true]));
+        }
+
+        return $this->redis;
+    }
+}