Added return value to TrackedNatsRequest::wait()

byrnedo · Nov 23, 2017 · 0534485 · 0534485
1 parent 55bd37f
commit 0534485
Show file tree

Hide file tree

Showing 5 changed files with 43 additions and 17 deletions.
diff --git a/README.md b/README.md
@@ -56,7 +56,10 @@ $c->connect();
 $r = $c->publish('special.subject', 'some serialized payload...');
 
 // optionally wait for the ack
-$r->wait();
+$gotAck = $r->wait();
+if (!$gotAck) {
+    ...
+}
 
 $c->close();
 
@@ -69,7 +72,10 @@ If publishing many messages at a time, you might at first do this:
 ```php
 foreach ($req as $data){
     $r = $c->publish(...);
-    $r->wait();
+    $gotAck = $r->wait();
+    if (!$gotAck) {
+        ...
+    }
 }
 ```
 

diff --git a/src/NatsStreaming/TrackedNatsRequest.php b/src/NatsStreaming/TrackedNatsRequest.php
@@ -71,6 +71,7 @@ private function dispatchCachedMessages()
         if ($cachedMsgs) {
             foreach ($cachedMsgs as $msg) {
                 $cb($msg);
+                $this->consumed = true;
             }
             // should only get 1 so get out
             return true;
@@ -79,16 +80,19 @@ private function dispatchCachedMessages()
         return false;
     }
 
+    public function gotAck() {
+        return $this->consumed > 0;
+    }
 
     public function wait()
     {
 
         if ($this->consumed) {
-            return;
+            return $this->gotAck();
         }
 
         if ($this->dispatchCachedMessages()) {
-            return;
+            return $this->gotAck();
         } else {
             $this->waiting = true;
 
@@ -102,8 +106,10 @@ public function wait()
         }
 
         $this->waiting = false;
+        return $this->gotAck();
     }
 
+
     /**
      * @return mixed
      */

diff --git a/tests/Integration/Commands/Producer b/tests/Integration/Commands/Producer
@@ -20,8 +20,11 @@ $subOpts = new \NatsStreaming\SubscriptionOptions();
 $count = 0;
 for(; $count < $numMessages;  $count++) {
     $ack = $con->publish($subject, 'foobar '.$count);
-    $ack->wait();
+    $gotAck = $ack->wait();
     echo "count : $count\n";
+    if (!$gotAck) {
+        error_log("ERROR: failed to get ACK");
+    }
 }
 
 

diff --git a/tests/Integration/MultipleConsumerTest.php b/tests/Integration/MultipleConsumerTest.php
@@ -12,7 +12,7 @@ class MultipleConsumerTest extends PHPUnit_Framework_TestCase
 
     public function setUp()
     {
-        parent::setUp(); // TODO: Change the autogenerated stub
+        parent::setUp();
         $this->phpCmd = str_replace("\n", '', `which php`);
         $this->logPrefix = $this->getName();
     }

diff --git a/tests/Unit/ConnectionTest.php b/tests/Unit/ConnectionTest.php
@@ -75,7 +75,8 @@ public function testPublish()
     {
         $this->c->reconnect();
         $r = $this->c->publish('foo', 'bar');
-        $r->wait();
+        $ack = $r->wait();
+        $this->assertTrue($ack);
         $count = $this->c->pubsCount();
         $this->assertInternalType('int', $count);
         $this->assertGreaterThan(0, $count);
@@ -121,7 +122,8 @@ public function testSubscribe()
         }
 
         foreach ($rs as $r) {
-            $r->wait();
+            $gotAck = $r->wait();
+            $this->assertTrue($gotAck);
         }
 
 
@@ -162,7 +164,8 @@ public function testGlobalWait()
         }
 
         foreach ($rs as $r) {
-            $r->wait();
+            $gotAck = $r->wait();
+            $this->assertTrue($gotAck);
         }
 
 
@@ -230,7 +233,8 @@ public function testMultipleSubscriptions()
         }
 
         foreach ($rs as $r) {
-            $r->wait(1);
+            $gotAck = $r->wait(1);
+            $this->assertTrue($gotAck);
         }
 
 
@@ -279,7 +283,8 @@ public function testDurableSubscription()
         }
 
         foreach ($rs as $r) {
-            $r->wait();
+            $gotAck = $r->wait();
+            $this->assertTrue($gotAck);
         }
 
         $sub->wait($toSend);
@@ -290,7 +295,8 @@ public function testDurableSubscription()
 
         $this->c->connect();
         $r = $this->c->publish($subject, 'foobarnew');
-        $r->wait();
+        $gotAck = $r->wait();
+        $this->assertTrue($gotAck);
 
         $subOptions = new \NatsStreaming\SubscriptionOptions();
 
@@ -367,7 +373,8 @@ public function testMultipleDurableSubscription()
 
         // quicker
         foreach ($rs as $r) {
-            $r->wait();
+            $gotAck = $r->wait();
+            $this->assertTrue($gotAck);
         }
 
         $sub1->wait($toSend);
@@ -381,7 +388,8 @@ public function testMultipleDurableSubscription()
         $this->c->connect();
 
         $r = $this->c->publish($subject, 'foobarnew');
-        $r->wait();
+        $gotAck = $r->wait();
+        $this->assertTrue($gotAck);
 
 
         $subOptions = new \NatsStreaming\SubscriptionOptions();
@@ -452,7 +460,8 @@ public function testQueueGroupSubscribe()
         }
 
         foreach ($rs as $r) {
-            $r->wait();
+            $gotAck = $r->wait();
+            $this->assertTrue($gotAck);
         }
 
         $sub->wait($toSend);
@@ -487,7 +496,8 @@ public function testUnsubscribe($close = false)
         }, $subOptions);
 
         $r = $this->c->publish($subject, 'foobar');
-        $r->wait();
+        $gotAck = $r->wait();
+        $this->assertTrue($gotAck);
 
         $sub->wait(1);
 
@@ -501,7 +511,8 @@ public function testUnsubscribe($close = false)
 
         $this->c->natsCon->setStreamTimeout(5);
         $r = $this->c->publish($subject, 'foobar');
-        $r->wait();
+        $gotAck = $r->wait();
+        $this->assertTrue($gotAck);
 
 
         $this->assertEquals(1, $got);