ack messages individually after callable returns successfully

Author: matthewgoslett

changelog.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 1.0.1 - ?
+
+* Ack messages individually after callable returns successfully
+
+* Initial release
+
 ## 1.0.0 - 2016-09-05
 
 * Initial release

src/GoogleCloudPubSubAdapter.php

@@ -44,31 +44,20 @@ public function subscribe($channel, callable $handler)
         $isSubscriptionLoopActive = true;
 
         while ($isSubscriptionLoopActive) {
-            $ackIds = [];
-            $payloads = [];
-
             $messages = $subscription->pull();
 
             foreach ($messages as $message) {
-                $ackIds[] = $message['ackId'];
-
                 // the cloud library base64 encodes messages
                 $payload = base64_decode($message['message']['data']);
                 $payload = Utils::unserializeMessagePayload($payload);
 
                 if ($payload === 'unsubscribe') {
                     $isSubscriptionLoopActive = false;
                 } else {
-                    $payloads[] = $payload;
+                    call_user_func($handler, $payload);
                 }
-            }
-
-            if (!empty($ackIds)) {
-                $subscription->acknowledgeBatch($ackIds);
-            }
 
-            foreach ($payloads as $payload) {
-                call_user_func($handler, $payload);
+                $subscription->acknowledge($message['ackId']);
             }
         }
     }

tests/GoogleCloudPubSubAdapterTest.php

@@ -101,14 +101,17 @@ public function testSubscribeWhenSubscriptionMustBeCreated()
         $subscription->shouldReceive('pull')
             ->once()
             ->andReturn($messageBatch1);
+        $subscription->shouldReceive('acknowledge')
+            ->with(1)
+            ->once();
+        $subscription->shouldReceive('acknowledge')
+            ->with(2)
+            ->once();
         $subscription->shouldReceive('pull')
             ->once()
             ->andReturn($messageBatch2);
-        $subscription->shouldReceive('acknowledgeBatch')
-            ->with([1, 2])
-            ->once();
-        $subscription->shouldReceive('acknowledgeBatch')
-            ->with([3])
+        $subscription->shouldReceive('acknowledge')
+            ->with(3)
             ->once();
 
         $topic = Mockery::mock(Topic::class);
@@ -173,14 +176,17 @@ public function testSubscribeWhenSubscriptionExists()
         $subscription->shouldReceive('pull')
             ->once()
             ->andReturn($messageBatch1);
+        $subscription->shouldReceive('acknowledge')
+            ->with(1)
+            ->once();
+        $subscription->shouldReceive('acknowledge')
+            ->with(2)
+            ->once();
         $subscription->shouldReceive('pull')
             ->once()
             ->andReturn($messageBatch2);
-        $subscription->shouldReceive('acknowledgeBatch')
-            ->with([1, 2])
-            ->once();
-        $subscription->shouldReceive('acknowledgeBatch')
-            ->with([3])
+        $subscription->shouldReceive('acknowledge')
+            ->with(3)
             ->once();
 
         $topic = Mockery::mock(Topic::class);