Add Gearman transport.

README.md

@@ -9,7 +9,7 @@ Features:
 
 * [Feature rich](docs/quick_tour.md).
 * [JMS](https://docs.oracle.com/javaee/7/api/javax/jms/package-summary.html) like transport [abstraction](https://github.com/php-enqueue/psr-queue).
-* Supports  transports [AMQP (RabbitMQ, ActiveMQ)](docs/transport/amqp.md), [Beanstalk (Pheanstalk)](docs/transport/pheanstalk.md),  [STOMP](docs/transport/stomp.md), [Amazon SQS](docs/transport/sqs.md), [Redis](docs/transport/redis.md), [Doctrine DBAL](docs/transport/dbal.md), [Filesystem](docs/transport/filesystem.md), [Null](docs/transport/null.md).
+* Supported  transports [AMQP](docs/transport/amqp.md) (RabbitMQ, ActiveMQ), [Beanstalk](docs/transport/pheanstalk.md),  [STOMP](docs/transport/stomp.md), [Amazon SQS](docs/transport/sqs.md), [Redis](docs/transport/redis.md), [Gearman](docs/transport/gearman.md), [Doctrine DBAL](docs/transport/dbal.md), [Filesystem](docs/transport/filesystem.md), [Null](docs/transport/null.md).
 * [Symfony bundle](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/bundle/quick_tour.md)
 * [Magento1 extension](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/magento/quick_tour.md)
 * [Message bus](http://www.enterpriseintegrationpatterns.com/patterns/messaging/MessageBus.html) support.

bin/test

@@ -24,6 +24,7 @@ waitForService rabbitmq 5672 50
 waitForService mysql 3306 50
 waitForService redis 6379 50
 waitForService beanstalkd 11300
+waitForService gearmand 4730
 
 php pkg/job-queue/Tests/Functional/app/console doctrine:database:create
 php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force

composer.json

@@ -14,6 +14,7 @@
     "enqueue/dbal": "*@dev",
     "enqueue/sqs": "*@dev",
     "enqueue/pheanstalk": "*@dev",
+    "enqueue/gearman": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
     "enqueue/simple-client": "*@dev",
@@ -90,6 +91,10 @@
       "type": "path",
       "url": "pkg/pheanstalk"
     },
+    {
+      "type": "path",
+      "url": "pkg/gearman"
+    },
     {
       "type": "path",
       "url": "pkg/simple-client"

docker-compose.yml

@@ -9,6 +9,7 @@ services:
       - mysql
       - redis
       - beanstalkd
+      - gearmand
     volumes:
       - './:/mqdev'
     environment:
@@ -34,6 +35,7 @@ services:
       - BEANSTALKD_HOST=beanstalkd
       - BEANSTALKD_PORT=11300
       - BEANSTALKD_DSN=beanstalk://beanstalkd:11300
+      - GEARMAN_DSN=gearman://gearmand:4730
 
   rabbitmq:
     image: enqueue/rabbitmq:latest
@@ -48,6 +50,9 @@ services:
   beanstalkd:
       image: 'schickling/beanstalkd'
 
+  gearmand:
+      image: 'artefactual/gearmand'
+
   redis:
     image: 'redis:3'
     ports:

docker/Dockerfile

@@ -1,16 +1,15 @@
-FROM ubuntu:16.04
+FROM formapro/nginx-php-fpm:latest-all-exts
 
 ## libs
 RUN set -x && \
     apt-get update && \
-    apt-get install -y --no-install-recommends wget curl openssl ca-certificates nano netcat && \
-    apt-get install -y --no-install-recommends php php-mysql php-redis php-curl php-intl php-mbstring php-zip php-mcrypt php-xdebug php-bcmath php-xml php-amqp
+    apt-get install -y --no-install-recommends wget curl openssl ca-certificates nano netcat
 
 ## confis
 
 # RUN rm -f /etc/php/7.0/cli/conf.d/*xdebug.ini
 
-COPY ./php/cli.ini /etc/php/7.0/cli/conf.d/1-dev_cli.ini
+COPY ./php/cli.ini /etc/php/7.1/cli/conf.d/1-dev_cli.ini
 COPY ./bin/dev_entrypoiny.sh /usr/local/bin/entrypoint.sh
 RUN chmod u+x /usr/local/bin/entrypoint.sh
 

docker/php/cli.ini

@@ -4,5 +4,3 @@ memory_limit = 2G
 max_execution_time=0
 date.timezone=UTC
 variables_order="EGPCS"
-
-extension=amqp.so

docs/transport/gearman.md

@@ -0,0 +1,84 @@
+# Gearman transport
+
+The transport uses [Gearman](http://gearman.org/) job manager. 
+The transport uses [Gearman PHP extension](http://php.net/manual/en/book.gearman.php) internally.      
+
+* [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)
+
+## Installation
+
+```bash
+$ composer require enqueue/gearman
+```
+
+
+## Create context
+
+```php
+<?php
+use Enqueue\Gearman\GearmanConnectionFactory;
+
+// connects to localhost:4730
+$factory = new GearmanConnectionFactory();
+
+// same as above
+$factory = new GearmanConnectionFactory('gearman://');
+
+// connects to example host and port 5555
+$factory = new GearmanConnectionFactory('gearman://example:5555');
+
+// same as above but configured by array
+$factory = new GearmanConnectionFactory([
+    'host' => 'example',
+    'port' => 5555
+]);
+```
+
+## Send message to topic
+
+```php
+<?php
+/** @var \Enqueue\Gearman\GearmanContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('aTopic');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooTopic, $message);
+```
+
+## Send message to queue 
+
+```php
+<?php
+/** @var \Enqueue\Gearman\GearmanContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooQueue, $message);
+```
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\Gearman\GearmanContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$consumer = $psrContext->createConsumer($fooQueue);
+
+$message = $consumer->receive(2000); // wait for 2 seconds
+
+$message = $consumer->receiveNoWait(); // fetch message or return null immediately 
+
+// process a message
+
+$consumer->acknowledge($message);
+// $consumer->reject($message);
+```
+
+[back to index](../index.md)

phpunit.xml.dist

@@ -53,6 +53,10 @@
             <directory>pkg/pheanstalk/Tests</directory>
         </testsuite>
 
+        <testsuite name="gearman transport">
+            <directory>pkg/gearman/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

pkg/enqueue/functions.php

@@ -6,7 +6,9 @@
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\Dbal\DbalConnectionFactory;
 use Enqueue\Fs\FsConnectionFactory;
+use Enqueue\Gearman\GearmanConnectionFactory;
 use Enqueue\Null\NullConnectionFactory;
+use Enqueue\Pheanstalk\PheanstalkConnectionFactory;
 use Enqueue\Psr\PsrConnectionFactory;
 use Enqueue\Psr\PsrContext;
 
@@ -48,6 +50,14 @@ function dsn_to_connection_factory($dsn)
         $map['pdo_sqlite'] = DbalConnectionFactory::class;
     }
 
+    if (class_exists(GearmanConnectionFactory::class)) {
+        $map['gearman'] = GearmanConnectionFactory::class;
+    }
+
+    if (class_exists(PheanstalkConnectionFactory::class)) {
+        $map['beanstalk'] = PheanstalkConnectionFactory::class;
+    }
+
     list($scheme) = explode('://', $dsn);
     if (false == $scheme || false === strpos($dsn, '://')) {
         throw new \LogicException(sprintf('The scheme could not be parsed from DSN "%s"', $dsn));

pkg/gearman/.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

pkg/gearman/GearmanConnectionFactory.php

@@ -0,0 +1,94 @@
+<?php
+
+namespace Enqueue\Gearman;
+
+use Enqueue\Psr\PsrConnectionFactory;
+
+class GearmanConnectionFactory implements PsrConnectionFactory
+{
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost with default settings.
+     *
+     * [
+     *     'host'  => 'localhost',
+     *     'port'  => 11300
+     * ]
+     *
+     * or
+     *
+     * gearman://host:port
+     *
+     * @param array|string $config
+     */
+    public function __construct($config = 'gearman://')
+    {
+        if (empty($config) || 'gearman://' === $config) {
+            $config = [];
+        } elseif (is_string($config)) {
+            $config = $this->parseDsn($config);
+        } elseif (is_array($config)) {
+        } else {
+            throw new \LogicException('The config must be either an array of options, a DSN string or null');
+        }
+
+        $this->config = array_replace($this->defaultConfig(), $config);
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return GearmanContext
+     */
+    public function createContext()
+    {
+        return new GearmanContext($this->config);
+    }
+
+    /**
+     * @param string $dsn
+     *
+     * @return array
+     */
+    private function parseDsn($dsn)
+    {
+        $dsnConfig = parse_url($dsn);
+        if (false === $dsnConfig) {
+            throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
+        }
+
+        $dsnConfig = array_replace([
+            'scheme' => null,
+            'host' => null,
+            'port' => null,
+            'user' => null,
+            'pass' => null,
+            'path' => null,
+            'query' => null,
+        ], $dsnConfig);
+
+        if ('gearman' !== $dsnConfig['scheme']) {
+            throw new \LogicException(sprintf('The given DSN scheme "%s" is not supported. Could be "gearman" only.', $dsnConfig['scheme']));
+        }
+
+        return [
+            'port' => $dsnConfig['port'],
+            'host' => $dsnConfig['host'],
+        ];
+    }
+
+    /**
+     * @return array
+     */
+    private function defaultConfig()
+    {
+        return [
+            'host' => \GEARMAN_DEFAULT_TCP_HOST,
+            'port' => \GEARMAN_DEFAULT_TCP_PORT,
+        ];
+    }
+}