add support for background batching messages (#5)

Author: matthewgoslett

.travis.yml

@@ -4,7 +4,6 @@ php:
   - 5.6
   - 7.0
   - 7.1
-  - hhvm
   - nightly
 
 before_script:

README.md

@@ -16,24 +16,6 @@ A Google Cloud adapter for the [php-pubsub](https://github.com/Superbalist/php-p
 composer require superbalist/php-pubsub-google-cloud
 ```
 
-## gRPC Support
-
-Google Cloud PHP v0.12.0 added support for communication over the gRPC protocol.
-
-> gRPC is great for high-performance, low-latency applications, and is highly recommended in cases where performance and latency are concerns.
-
-The library will automatically choose gRPC over REST if all dependencies are installed.
-* [gRPC PECL extension](https://pecl.php.net/package/gRPC)
-* [google/proto-client-php composer package](https://github.com/googleapis/gax-php)
-* [googleapis/proto-client-php composer package](https://github.com/googleapis/proto-client-php)
-
-```bash
-pecl install grpc
-
-composer require google/gax
-composer require google/proto-client-php ^0.10.0
-```
-
 ## Usage
 
 ```php
@@ -66,6 +48,44 @@ $adapter->publish('my_channel', 1);
 $adapter->publish('my_channel', false);
 ```
 
+## gRPC Support
+
+Google Cloud PHP v0.12.0 added support for communication over the gRPC protocol.
+
+> gRPC is great for high-performance, low-latency applications, and is highly recommended in cases where performance and latency are concerns.
+
+The library will automatically choose gRPC over REST if all dependencies are installed.
+* [gRPC PECL extension](https://pecl.php.net/package/gRPC)
+* [google/proto-client-php composer package](https://github.com/googleapis/gax-php)
+* [googleapis/proto-client-php composer package](https://github.com/googleapis/proto-client-php)
+
+```bash
+pecl install grpc
+
+composer require google/gax
+composer require google/proto-client
+```
+
+## Background Batch Message Support
+
+Google Cloud v0.33.0 added support for queueing messages and publishing in the background.  This is available in
+version 5+ of this package which requires a min version of google/cloud ^0.33.0.
+
+You can enable background batch messaging by setting `$backgroundBatching` to `true` when constructing the
+`GoogleCloudPubSubAdapter` or by calling `setBackgroundBatching(true)` on an existing adapter.
+
+If the [semaphore](http://php.net/manual/en/book.sem.php) and [pcntl](http://php.net/manual/en/book.pcntl.php) PHP extensions are
+enabled AND the `IS_BATCH_DAEMON_RUNNING` ENV var is set to `true`, the library will queue messages to be published by
+the [Batch Daemon](https://github.com/GoogleCloudPlatform/google-cloud-php/blob/master/src/Core/Batch/BatchDaemon.php).
+The Batch Daemon needs to be manually run as a long-lived background process.
+
+For all other cases, messages will be queued in memory and will be published before the script terminates using a
+vendor registered shutdown handler.
+
+**Please Note**
+
+This is marked by google/cloud as an experimental feature & may change before release in backwards-incompatible ways.
+
 ## Examples
 
 The library comes with [examples](examples) for the adapter and a [Dockerfile](Dockerfile) for

changelog.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 5.0.0 - 2017-07-25
+
+* Add support for using Google Cloud batch requests (aka background daemon)
+
 ## 4.0.1 - 2017-07-21
 
 * Allow for google/cloud ^0.29.0|^0.30.0|^0.31.0|^0.32.0|^0.33.0|^0.34.0|^0.35.0

composer.json

@@ -11,7 +11,7 @@
   "require": {
     "php": ">=5.6.0",
     "superbalist/php-pubsub": "^2.0",
-    "google/cloud": "^0.11.0|^0.12.0|^0.13.0|^0.20.0|^0.21.0|^0.22.0|^0.23.0|^0.24.0|^0.25.0|^0.26.0|^0.27.0|^0.28.0|^0.29.0|^0.30.0|^0.31.0|^0.32.0|^0.33.0|^0.34.0|^0.35.0"
+    "google/cloud": "^0.33.0|^0.34.0|^0.35.0"
   },
   "autoload": {
     "psr-4": {

src/GoogleCloudPubSubAdapter.php

@@ -29,22 +29,30 @@ class GoogleCloudPubSubAdapter implements PubSubAdapterInterface
      */
     protected $autoCreateSubscriptions;
 
+    /**
+     * @var bool
+     */
+    protected $backgroundBatching;
+
     /**
      * @param PubSubClient $client
      * @param string $clientIdentifier
      * @param bool $autoCreateTopics
      * @param bool $autoCreateSubscriptions
+     * @param bool $backgroundBatching
      */
     public function __construct(
         PubSubClient $client,
         $clientIdentifier = null,
         $autoCreateTopics = true,
-        $autoCreateSubscriptions = true
+        $autoCreateSubscriptions = true,
+        $backgroundBatching = false
     ) {
         $this->client = $client;
         $this->clientIdentifier = $clientIdentifier;
         $this->autoCreateTopics = $autoCreateTopics;
         $this->autoCreateSubscriptions = $autoCreateSubscriptions;
+        $this->backgroundBatching = $backgroundBatching;
     }
 
     /**
@@ -123,6 +131,35 @@ public function areSubscriptionsAutoCreated()
         return $this->autoCreateSubscriptions;
     }
 
+    /**
+     * Set whether or not background batching is enabled.
+     *
+     * This is available from Google Cloud 0.33+ - https://github.com/GoogleCloudPlatform/google-cloud-php/releases/tag/v0.33.0
+     *
+     * If the http://php.net/manual/en/book.sem.php and http://php.net/manual/en/book.pcntl.php extensions are enabled
+     * AND the IS_BATCH_DAEMON_RUNNING ENV var is set to true, the library will queue messages to be published by the
+     * Batch Daemon (https://github.com/GoogleCloudPlatform/google-cloud-php/blob/master/src/Core/Batch/BatchDaemon.php)
+     *
+     * For all other cases, messages will be queued in memory and will be published before the script terminates using
+     * a vendor registered shutdown handler.
+     *
+     * @param bool $backgroundBatching
+     */
+    public function setBackgroundBatching($backgroundBatching)
+    {
+        $this->backgroundBatching = $backgroundBatching;
+    }
+
+    /**
+     * Check whether or not background batching is enabled.
+     *
+     * @return bool
+     */
+    public function isBackgroundBatchingEnabled()
+    {
+        return $this->backgroundBatching;
+    }
+
     /**
      * Subscribe a handler to a channel.
      *
@@ -166,7 +203,13 @@ public function subscribe($channel, callable $handler)
     public function publish($channel, $message)
     {
         $topic = $this->getTopicForChannel($channel);
-        $topic->publish(['data' => Utils::serializeMessage($message)]);
+        $payload = Utils::serializeMessage($message);
+
+        if ($this->backgroundBatching) {
+            $topic->batchPublisher()->publish(['data' => $payload]);
+        } else {
+            $topic->publish(['data' => $payload]);
+        }
     }
 
     /**
@@ -181,7 +224,15 @@ public function publishBatch($channel, array $messages)
         $messages = array_map(function ($message) {
             return ['data' => Utils::serializeMessage($message)];
         }, $messages);
-        $topic->publishBatch($messages);
+
+        if ($this->backgroundBatching) {
+            $batchPublisher = $topic->batchPublisher();
+            foreach ($messages as $message) {
+                $batchPublisher->publish($message);
+            }
+        } else {
+            $topic->publishBatch($messages);
+        }
     }
 
     /**

tests/GoogleCloudPubSubAdapterTest.php

@@ -2,6 +2,7 @@
 
 namespace Tests;
 
+use Google\Cloud\PubSub\BatchPublisher;
 use Google\Cloud\PubSub\Message;
 use Google\Cloud\PubSub\PubSubClient;
 use Google\Cloud\PubSub\Subscription;
@@ -49,6 +50,19 @@ public function testGetSetAutoCreateSubscriptions()
         $this->assertFalse($adapter->areSubscriptionsAutoCreated());
     }
 
+    public function testGetSetBackgroundBatching()
+    {
+        $client = Mockery::mock(PubSubClient::class);
+        $adapter = new GoogleCloudPubSubAdapter($client);
+        $this->assertFalse($adapter->isBackgroundBatchingEnabled());
+
+        $adapter->setBackgroundBatching(true);
+        $this->assertTrue($adapter->isBackgroundBatchingEnabled());
+
+        $adapter = new GoogleCloudPubSubAdapter($client, null, true, true, true);
+        $this->assertTrue($adapter->isBackgroundBatchingEnabled());
+    }
+
     public function testPublishWhenTopicMustBeCreated()
     {
         $topic = Mockery::mock(Topic::class);
@@ -74,6 +88,39 @@ public function testPublishWhenTopicMustBeCreated()
         $adapter->publish('channel_name', ['hello' => 'world']);
     }
 
+    public function testPublishWhenTopicMustBeCreatedAndBackgroundBatchingIsEnabled()
+    {
+        $batchPublisher = Mockery::mock(BatchPublisher::class);
+
+        $topic = Mockery::mock(Topic::class);
+        $topic->shouldReceive('exists')
+            ->once()
+            ->andReturn(false);
+        $topic->shouldReceive('create')
+            ->once();
+
+        $topic->shouldReceive('batchPublisher')
+            ->once()
+            ->andReturn($batchPublisher);
+
+        $batchPublisher->shouldReceive('publish')
+            ->with([
+                'data' => '{"hello":"world"}',
+            ])
+            ->once();
+
+        $client = Mockery::mock(PubSubClient::class);
+        $client->shouldReceive('topic')
+            ->with('channel_name')
+            ->once()
+            ->andReturn($topic);
+
+        $adapter = new GoogleCloudPubSubAdapter($client);
+        $adapter->setBackgroundBatching(true);
+
+        $adapter->publish('channel_name', ['hello' => 'world']);
+    }
+
     public function testPublishWhenTopicExists()
     {
         $topic = Mockery::mock(Topic::class);
@@ -98,6 +145,37 @@ public function testPublishWhenTopicExists()
         $adapter->publish('channel_name', ['hello' => 'world']);
     }
 
+    public function testPublishWhenTopicExistsAndBackgroundBatchingIsEnabled()
+    {
+        $batchPublisher = Mockery::mock(BatchPublisher::class);
+
+        $topic = Mockery::mock(Topic::class);
+        $topic->shouldReceive('exists')
+            ->once()
+            ->andReturn(true);
+        $topic->shouldNotHaveReceived('create');
+        $topic->shouldReceive('batchPublisher')
+            ->once()
+            ->andReturn($batchPublisher);
+
+        $batchPublisher->shouldReceive('publish')
+            ->with([
+                'data' => '{"hello":"world"}',
+            ])
+            ->once();
+
+        $client = Mockery::mock(PubSubClient::class);
+        $client->shouldReceive('topic')
+            ->with('channel_name')
+            ->once()
+            ->andReturn($topic);
+
+        $adapter = new GoogleCloudPubSubAdapter($client);
+        $adapter->setBackgroundBatching(true);
+
+        $adapter->publish('channel_name', ['hello' => 'world']);
+    }
+
     public function testPublishWhenAutoTopicCreationIsDisabled()
     {
         $topic = Mockery::mock(Topic::class);
@@ -120,6 +198,35 @@ public function testPublishWhenAutoTopicCreationIsDisabled()
         $adapter->publish('channel_name', ['hello' => 'world']);
     }
 
+    public function testPublishWhenAutoTopicCreationIsDisabledAndBackgroundBatchingIsEnabled()
+    {
+        $batchPublisher = Mockery::mock(BatchPublisher::class);
+
+        $topic = Mockery::mock(Topic::class);
+        $topic->shouldNotHaveReceived('exists');
+        $topic->shouldNotHaveReceived('create');
+        $topic->shouldReceive('batchPublisher')
+            ->once()
+            ->andReturn($batchPublisher);
+
+        $batchPublisher->shouldReceive('publish')
+            ->with([
+                'data' => '{"hello":"world"}',
+            ])
+            ->once();
+
+        $client = Mockery::mock(PubSubClient::class);
+        $client->shouldReceive('topic')
+            ->with('channel_name')
+            ->once()
+            ->andReturn($topic);
+
+        $adapter = new GoogleCloudPubSubAdapter($client, null, false);
+        $adapter->setBackgroundBatching(true);
+
+        $adapter->publish('channel_name', ['hello' => 'world']);
+    }
+
     public function testPublishBatch()
     {
         $topic = Mockery::mock(Topic::class);
@@ -148,6 +255,45 @@ public function testPublishBatch()
         $adapter->publishBatch('channel_name', $messages);
     }
 
+    public function testPublishBatchWhenBackgroundBatchingIsEnabled()
+    {
+        $batchPublisher = Mockery::mock(BatchPublisher::class);
+
+        $topic = Mockery::mock(Topic::class);
+        $topic->shouldReceive('exists')
+            ->once()
+            ->andReturn(true);
+        $topic->shouldReceive('batchPublisher')
+            ->once()
+            ->andReturn($batchPublisher);
+
+        $batchPublisher->shouldReceive('publish')
+            ->with([
+                'data' => '{"hello":"world"}',
+            ])
+            ->once();
+        $batchPublisher->shouldReceive('publish')
+            ->with([
+                'data' => '"booo!"',
+            ])
+            ->once();
+
+        $client = Mockery::mock(PubSubClient::class);
+        $client->shouldReceive('topic')
+            ->with('channel_name')
+            ->once()
+            ->andReturn($topic);
+
+        $adapter = new GoogleCloudPubSubAdapter($client);
+        $adapter->setBackgroundBatching(true);
+
+        $messages = [
+            ['hello' => 'world'],
+            'booo!',
+        ];
+        $adapter->publishBatch('channel_name', $messages);
+    }
+
     public function testSubscribeWhenSubscriptionMustBeCreated()
     {
         $message1 = new Message(['data' => '{"hello":"world"}'], ['ackId' => 1]);