Merge pull request #55 from php-enqueue/redis-transport

Redis transport.

Author: makasim

Diff for: bin/test

@@ -22,6 +22,7 @@ function waitForService()
 
 waitForService rabbitmq 5672 50
 waitForService mysql 3306 50
+waitForService redis 6379 50
 
 php pkg/job-queue/Tests/Functional/app/console doctrine:database:create
 php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force

Diff for: composer.json

@@ -8,14 +8,15 @@
     "enqueue/enqueue": "*@dev",
     "enqueue/stomp": "*@dev",
     "enqueue/amqp-ext": "*@dev",
+    "enqueue/redis": "*@dev",
     "enqueue/fs": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
-    "enqueue/test": "*@dev"
-  },
-  "require-dev": {
+    "enqueue/test": "*@dev",
+
     "phpunit/phpunit": "^5",
     "doctrine/doctrine-bundle": "~1.2",
+    "predis/predis": "^1.1",
     "symfony/monolog-bundle": "^2.8|^3",
     "symfony/browser-kit": "^2.8|^3",
     "symfony/expression-language": "^2.8|^3",
@@ -46,6 +47,10 @@
       "type": "path",
       "url": "pkg/amqp-ext"
     },
+    {
+      "type": "path",
+      "url": "pkg/redis"
+    },
     {
       "type": "path",
       "url": "pkg/enqueue-bundle"

Diff for: docker-compose.yml

@@ -6,8 +6,9 @@ services:
     depends_on:
       - rabbitmq
       - mysql
+      - redis
     volumes:
-      - ./:/mqdev
+      - './:/mqdev'
     environment:
       - SYMFONY__RABBITMQ__HOST=rabbitmq
       - SYMFONY__RABBITMQ__USER=guest
@@ -21,6 +22,8 @@ services:
       - SYMFONY__DB__NAME=mqdev
       - SYMFONY__DB__USER=root
       - SYMFONY__DB__PASSWORD=rootpass
+      - SYMFONY__REDIS__HOST=redis
+      - SYMFONY__REDIS__PORT=6379
 
   rabbitmq:
     image: enqueue/rabbitmq:latest
@@ -29,6 +32,11 @@ services:
       - RABBITMQ_DEFAULT_USER=guest
       - RABBITMQ_DEFAULT_PASS=guest
       - RABBITMQ_DEFAULT_VHOST=mqdev
+  redis:
+    image: 'redis:3'
+    ports:
+      - "6379:6379"
+
   mysql:
     image: mariadb:10
     volumes:

Diff for: docs/filesystem_transport.md

@@ -3,13 +3,10 @@
 Use files on local filesystem as queues. 
 It creates a file per queue\topic. 
 A message is a line inside the file.
-**Limitations** It works only in auto ack mode. Local by nature therefor messages are not visible on other servers.  
+**Limitations** It works only in auto ack mode hence If consumer crashes the message is lost. Local by nature therefor messages are not visible on other servers.  
 
 * [Installation](#installation)
 * [Create context](#create-context)
-* [Declare topic](#declare-topic)
-* [Declare queue](#decalre-queue)
-* [Bind queue to topic](#bind-queue-to-topic)
 * [Send message to topic](#send-message-to-topic)
 * [Send message to queue](#send-message-to-queue)
 * [Consume message](#consume-message)

Diff for: docs/index.md

@@ -2,8 +2,9 @@
 
 * [Quick tour](quick_tour.md)
 * Transports
-    - [Amqp](amqp_transport.md)
-    - [Stomp](stomp_transport.md)
+    - [Amqp (RabbitMQ, ActiveMQ)](amqp_transport.md)
+    - [Stomp (RabbitMQ, ActiveMQ)](stomp_transport.md)
+    - [Redis](redis_transport.md)
     - [Filesystem](filesystem_transport.md)
     - [Null](null_transport.md)
 * Consumption

Diff for: docs/redis_transport.md

@@ -0,0 +1,125 @@
+# Redis transport
+
+The transport uses [Redis](https://redis.io/) as a message broker. 
+It creates a collection (a queue or topic) there. Pushes messages to the tail of the collection and pops from the head.
+The transport works with [phpredis](https://github.com/phpredis/phpredis) php extension or [predis](https://github.com/nrk/predis) library. 
+Make sure you installed either of them 
+ 
+**Limitations** It works only in auto ack mode hence If consumer crashes the message is lost.  
+
+* [Installation](#installation)
+* [Create context](#create-context)
+* [Send message to topic](#send-message-to-topic)
+* [Send message to queue](#send-message-to-queue)
+* [Consume message](#consume-message)
+* [Delete queue (purge messages)](#delete-queue-purge-messages)
+* [Delete topic (purge messages)](#delete-topic-purge-messages)
+
+## Installation
+
+* With php redis extension:
+
+```bash
+$ apt-get install php-redis
+$ composer require enqueue/redis
+```
+
+* With predis library:
+
+```bash
+$ composer require enqueue/redis predis/predis:^1
+```
+
+## Create context
+
+* With php redis extension:
+
+```php
+<?php
+use Enqueue\Redis\RedisConnectionFactory;
+
+$connectionFactory = new RedisConnectionFactory([
+    'host' => 'localhost',
+    'port' => 6379,
+    'vendor' => 'phpredis',
+]);
+
+$psrContext = $connectionFactory->createContext();
+```
+
+* With predis library:
+
+```php
+<?php
+use Enqueue\Redis\RedisConnectionFactory;
+
+$connectionFactory = new RedisConnectionFactory([
+    'host' => 'localhost',
+    'port' => 6379,
+    'vendor' => 'predis',
+]);
+
+$psrContext = $connectionFactory->createContext();
+```
+
+## Send message to topic
+
+```php
+<?php
+/** @var \Enqueue\Redis\RedisContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('aTopic');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooTopic, $message);
+```
+
+## Send message to queue 
+
+```php
+<?php
+/** @var \Enqueue\Redis\RedisContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooQueue, $message);
+```
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\Redis\RedisContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$consumer = $psrContext->createConsumer($fooQueue);
+
+$message = $consumer->receive();
+
+// process a message
+```
+
+## Delete queue (purge messages):
+
+```php
+<?php
+/** @var \Enqueue\Redis\RedisContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+
+$psrContext->deleteQueue($fooQueue);
+```
+
+## Delete topic (purge messages):
+
+```php
+<?php
+/** @var \Enqueue\Redis\RedisContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('aTopic');
+
+$psrContext->deleteTopic($fooTopic);
+```
+
+[back to index](index.md)

Diff for: phpunit.xml.dist

@@ -29,10 +29,14 @@
             <directory>pkg/amqp-ext/Tests</directory>
         </testsuite>
 
-        <testsuite name="fs">
+        <testsuite name="fs transport">
             <directory>pkg/fs/Tests</directory>
         </testsuite>
 
+        <testsuite name="redis transport">
+            <directory>pkg/redis/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

Diff for: pkg/amqp-ext/AmqpConnectionFactory.php

@@ -26,7 +26,8 @@ class AmqpConnectionFactory implements PsrConnectionFactory
      *     'read_timeout'  => Timeout in for income activity. Note: 0 or greater seconds. May be fractional.
      *     'write_timeout' => Timeout in for outcome activity. Note: 0 or greater seconds. May be fractional.
      *     'connect_timeout' => Connection timeout. Note: 0 or greater seconds. May be fractional.
-     *     'persisted' => bool
+     *     '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

Diff for: pkg/enqueue/Rpc/Promise.php

@@ -50,6 +50,7 @@ public function getMessage()
 
                     return $message;
                 }
+
                 $this->consumer->reject($message, true);
             }
         }

Diff for: pkg/fs/FsConsumer.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Fs;
 
+use Enqueue\Psr\InvalidMessageException;
 use Enqueue\Psr\PsrConsumer;
 use Enqueue\Psr\PsrMessage;
 
@@ -118,6 +119,8 @@ public function acknowledge(PsrMessage $message)
      */
     public function reject(PsrMessage $message, $requeue = false)
     {
+        InvalidMessageException::assertMessageInstanceOf($message, FsMessage::class);
+
         // do nothing on reject. fs transport always works in auto ack mode
 
         if ($requeue) {

Diff for: pkg/redis/.gitignore

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

Diff for: pkg/redis/.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

Diff for: pkg/redis/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.