Google Pub/Sub

README.md

@@ -14,6 +14,7 @@ Features:
     * [Beanstalk](docs/transport/pheanstalk.md)
     * [STOMP](docs/transport/stomp.md)
     * [Amazon SQS](docs/transport/sqs.md)
+    * [Google PubSub](docs/transport/gps.md)
     * [Kafka](docs/transport/kafka.md)
     * [Redis](docs/transport/redis.md)
     * [Gearman](docs/transport/gearman.md)

composer.json

@@ -20,6 +20,7 @@
     "enqueue/gearman": "*@dev",
     "enqueue/rdkafka": "*@dev",
     "kwn/php-rdkafka-stubs": "^1.0.2",
+    "enqueue/gps": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
     "enqueue/simple-client": "*@dev",
@@ -129,6 +130,10 @@
       "type": "path",
       "url": "pkg/rdkafka"
     },
+    {
+      "type": "path",
+      "url": "pkg/gps"
+    },
     {
       "type": "path",
       "url": "pkg/simple-client"

docker-compose.yml

@@ -12,6 +12,7 @@ services:
       - gearmand
       - kafka
       - zookeeper
+      - google-pubsub
     volumes:
       - './:/mqdev'
     environment:
@@ -40,6 +41,8 @@ services:
       - GEARMAN_DSN=gearman://gearmand:4730
       - RDKAFKA_HOST=kafka
       - RDKAFKA_PORT=9092
+      - PUBSUB_EMULATOR_HOST=http://google-pubsub:8085
+      - GCLOUD_PROJECT=mqdev
 
   rabbitmq:
     image: enqueue/rabbitmq:latest
@@ -89,6 +92,10 @@ services:
     volumes:
       - '/var/run/docker.sock:/var/run/docker.sock'
 
+  google-pubsub:
+      image: 'google/cloud-sdk:latest'
+      entrypoint: 'gcloud beta emulators pubsub start --host-port=0.0.0.0:8085'
+
 volumes:
   mysql-data:
     driver: local

docs/index.md

@@ -4,6 +4,7 @@
 * [Transports](#transports)
     - Amqp based on [the ext](transport/amqp.md), [bunny](transport/amqp_bunny.md), [the lib](transport/amqp_lib.md)
     - [Amazon SQS](transport/sqs.md)
+    - [Google PubSub](transport/gps.md)
     - [Beanstalk (Pheanstalk)](transport/pheanstalk.md)
     - [Gearman](transport/gearman.md)
     - [Kafka](transport/kafka.md)

docs/transport/gps.md

@@ -0,0 +1,74 @@
+# Google Pub Sub transport
+
+A transport for [Google Pub Sub](https://cloud.google.com/pubsub/docs/) cloud MQ.
+It uses internally official google sdk library [google/cloud-pubsub](https://packagist.org/packages/google/cloud-pubsub) 
+
+* [Installation](#installation)
+* [Create context](#create-context)
+* [Send message to topic](#send-message-to-topic)
+* [Consume message](#consume-message)
+
+## Installation
+
+```bash
+$ composer require enqueue/gps
+```
+
+## Create context
+
+To enable the Google Cloud Pub/Sub Emulator, set the `PUBSUB_EMULATOR_HOST` environment variable. 
+There is a handy docker container [google/cloud-sdk](https://hub.docker.com/r/google/cloud-sdk/).  
+
+```php
+<?php
+use Enqueue\Gps\GpsConnectionFactory;
+
+putenv('PUBSUB_EMULATOR_HOST=http://localhost:8900');
+
+$connectionFactory = new GpsConnectionFactory();
+
+$psrContext = $connectionFactory->createContext();
+```
+
+## Send message to topic
+
+Before you can send message you have to declare a topic. 
+The operation creates a topic on a broker side. 
+Google allows messages to be sent only to topic. 
+
+```php
+<?php
+/** @var \Enqueue\Gps\GpsContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('foo');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->declareTopic($fooTopic);
+
+$psrContext->createProducer()->send($fooTopic, $message);
+```
+
+## Consume message:
+
+Before you can consume message you have to subscribe a queue to the topic. 
+Google does not allow consuming message from the topic directly. 
+
+```php
+<?php
+/** @var \Enqueue\Gps\GpsContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('foo');
+$fooQueue = $psrContext->createQueue('foo');
+
+$psrContext->subscribe($fooTopic, $fooQueue);
+
+$consumer = $psrContext->createConsumer($fooQueue);
+$message = $consumer->receive();
+
+// process a message
+
+$consumer->acknowledge($message);
+// $consumer->reject($message);
+```
+
+[back to index](../index.md)

phpunit.xml.dist

@@ -73,6 +73,10 @@
             <directory>pkg/rdkafka/Tests</directory>
         </testsuite>
 
+        <testsuite name="Google Pub/Sub transport">
+            <directory>pkg/gps/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

pkg/gps/.gitignore

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

pkg/gps/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 10
+
+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/gps/Client/GpsDriver.php

@@ -0,0 +1,182 @@
+<?php
+
+namespace Enqueue\Gps\Client;
+
+use Enqueue\Client\Config;
+use Enqueue\Client\DriverInterface;
+use Enqueue\Client\Message;
+use Enqueue\Client\Meta\QueueMetaRegistry;
+use Enqueue\Gps\GpsContext;
+use Enqueue\Gps\GpsMessage;
+use Enqueue\Gps\GpsQueue;
+use Enqueue\Gps\GpsTopic;
+use Interop\Queue\PsrMessage;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+class GpsDriver implements DriverInterface
+{
+    /**
+     * @var GpsContext
+     */
+    private $context;
+
+    /**
+     * @var Config
+     */
+    private $config;
+
+    /**
+     * @var QueueMetaRegistry
+     */
+    private $queueMetaRegistry;
+
+    /**
+     * @param GpsContext        $context
+     * @param Config            $config
+     * @param QueueMetaRegistry $queueMetaRegistry
+     */
+    public function __construct(GpsContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
+    {
+        $this->context = $context;
+        $this->config = $config;
+        $this->queueMetaRegistry = $queueMetaRegistry;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function sendToRouter(Message $message)
+    {
+        if (false == $message->getProperty(Config::PARAMETER_TOPIC_NAME)) {
+            throw new \LogicException('Topic name parameter is required but is not set');
+        }
+
+        $topic = $this->createRouterTopic();
+        $transportMessage = $this->createTransportMessage($message);
+
+        $this->context->createProducer()->send($topic, $transportMessage);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function sendToProcessor(Message $message)
+    {
+        if (false == $message->getProperty(Config::PARAMETER_PROCESSOR_NAME)) {
+            throw new \LogicException('Processor name parameter is required but is not set');
+        }
+
+        if (false == $queueName = $message->getProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME)) {
+            throw new \LogicException('Queue name parameter is required but is not set');
+        }
+
+        $transportMessage = $this->createTransportMessage($message);
+        $destination = $this->context->createTopic(
+            $this->queueMetaRegistry->getQueueMeta($queueName)->getTransportName())
+        ;
+
+        $this->context->createProducer()->send($destination, $transportMessage);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function setupBroker(LoggerInterface $logger = null)
+    {
+        $logger = $logger ?: new NullLogger();
+        $log = function ($text, ...$args) use ($logger) {
+            $logger->debug(sprintf('[GpsDriver] '.$text, ...$args));
+        };
+
+        // setup router
+        $routerTopic = $this->createRouterTopic();
+        $routerQueue = $this->createQueue($this->config->getRouterQueueName());
+
+        $log('Subscribe router topic to queue: %s -> %s', $routerTopic->getTopicName(), $routerQueue->getQueueName());
+        $this->context->subscribe($routerTopic, $routerQueue);
+
+        // setup queues
+        foreach ($this->queueMetaRegistry->getQueuesMeta() as $meta) {
+            $topic = $this->context->createTopic($meta->getTransportName());
+            $queue = $this->context->createQueue($meta->getTransportName());
+
+            $log('Subscribe processor topic to queue: %s -> %s', $topic->getTopicName(), $queue->getQueueName());
+            $this->context->subscribe($topic, $queue);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return GpsQueue
+     */
+    public function createQueue($queueName)
+    {
+        $transportName = $this->queueMetaRegistry->getQueueMeta($queueName)->getTransportName();
+
+        return $this->context->createQueue($transportName);
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return GpsMessage
+     */
+    public function createTransportMessage(Message $message)
+    {
+        $headers = $message->getHeaders();
+        $properties = $message->getProperties();
+
+        $transportMessage = $this->context->createMessage();
+        $transportMessage->setBody($message->getBody());
+        $transportMessage->setHeaders($headers);
+        $transportMessage->setProperties($properties);
+        $transportMessage->setMessageId($message->getMessageId());
+        $transportMessage->setTimestamp($message->getTimestamp());
+        $transportMessage->setReplyTo($message->getReplyTo());
+        $transportMessage->setCorrelationId($message->getCorrelationId());
+
+        return $transportMessage;
+    }
+
+    /**
+     * @param GpsMessage $message
+     *
+     * {@inheritdoc}
+     */
+    public function createClientMessage(PsrMessage $message)
+    {
+        $clientMessage = new Message();
+
+        $clientMessage->setBody($message->getBody());
+        $clientMessage->setHeaders($message->getHeaders());
+        $clientMessage->setProperties($message->getProperties());
+        $clientMessage->setMessageId($message->getMessageId());
+        $clientMessage->setTimestamp($message->getTimestamp());
+        $clientMessage->setReplyTo($message->getReplyTo());
+        $clientMessage->setCorrelationId($message->getCorrelationId());
+
+        return $clientMessage;
+    }
+
+    /**
+     * @return Config
+     */
+    public function getConfig()
+    {
+        return $this->config;
+    }
+
+    /**
+     * @return GpsTopic
+     */
+    private function createRouterTopic()
+    {
+        $topic = $this->context->createTopic(
+            $this->config->createTransportRouterTopicName($this->config->getRouterTopicName())
+        );
+
+        return $topic;
+    }
+}