RPC Deletes Reply Queue After Receive Message

docs/client/rpc_call.md

@@ -43,7 +43,7 @@ $promises[] = $rpcClient->callAsync('greeting_topic', 'Hi Thomas!', 5);
 
 $replyMessages = [];
 foreach ($promises as $promise) {
-    $replyMessages[] = $promise->getMessage();
+    $replyMessages[] = $promise->receive();
 }
 ```
 

docs/quick_tour.md

@@ -124,7 +124,7 @@ $message = $psrContext->createMessage('Hi there!');
 $rpcClient = new RpcClient($psrContext);
 
 $promise = $rpcClient->callAsync($queue, $message, 1);
-$replyMessage = $promise->getMessage();
+$replyMessage = $promise->receive();
 ```
 
 There is also extensions for the consumption component. 

pkg/enqueue/Client/RpcClient.php

@@ -5,6 +5,7 @@
 use Enqueue\Psr\PsrContext;
 use Enqueue\Psr\PsrMessage;
 use Enqueue\Rpc\Promise;
+use Enqueue\Rpc\TimeoutException;
 use Enqueue\Util\UUID;
 
 class RpcClient
@@ -38,7 +39,7 @@ public function __construct(ProducerInterface $producer, PsrContext $context)
      */
     public function call($topic, $message, $timeout)
     {
-        return $this->callAsync($topic, $message, $timeout)->getMessage();
+        return $this->callAsync($topic, $message, $timeout)->receive();
     }
 
     /**
@@ -62,9 +63,11 @@ public function callAsync($topic, $message, $timeout)
 
         if ($message->getReplyTo()) {
             $replyQueue = $this->context->createQueue($message->getReplyTo());
+            $deleteReplyQueue = false;
         } else {
             $replyQueue = $this->context->createTemporaryQueue();
             $message->setReplyTo($replyQueue->getQueueName());
+            $deleteReplyQueue = true;
         }
 
         if (false == $message->getCorrelationId()) {
@@ -73,10 +76,54 @@ public function callAsync($topic, $message, $timeout)
 
         $this->producer->send($topic, $message);
 
-        return new Promise(
-            $this->context->createConsumer($replyQueue),
-            $message->getCorrelationId(),
-            $timeout
-        );
+        $correlationId = $message->getCorrelationId();
+
+        $receive = function () use ($replyQueue, $timeout, $correlationId) {
+            $endTime = time() + ((int) ($timeout / 1000));
+            $consumer = $this->context->createConsumer($replyQueue);
+
+            do {
+                if ($message = $consumer->receive($timeout)) {
+                    if ($message->getCorrelationId() === $correlationId) {
+                        $consumer->acknowledge($message);
+
+                        return $message;
+                    }
+
+                    $consumer->reject($message, true);
+                }
+            } while (time() < $endTime);
+
+            throw TimeoutException::create($timeout, $correlationId);
+        };
+
+        $receiveNoWait = function () use ($replyQueue, $correlationId) {
+            static $consumer;
+
+            if (null === $consumer) {
+                $consumer = $this->context->createConsumer($replyQueue);
+            }
+
+            if ($message = $consumer->receiveNoWait()) {
+                if ($message->getCorrelationId() === $correlationId) {
+                    $consumer->acknowledge($message);
+
+                    return $message;
+                }
+
+                $consumer->reject($message, true);
+            }
+        };
+
+        $finally = function (Promise $promise) use ($replyQueue) {
+            if ($promise->isDeleteReplyQueue() && method_exists($this->context, 'deleteQueue')) {
+                $this->context->deleteQueue($replyQueue);
+            }
+        };
+
+        $promise = new Promise($receive, $receiveNoWait, $finally);
+        $promise->setDeleteReplyQueue($deleteReplyQueue);
+
+        return $promise;
     }
 }

pkg/enqueue/Rpc/Promise.php

@@ -2,75 +2,135 @@
 
 namespace Enqueue\Rpc;
 
-use Enqueue\Psr\PsrConsumer;
 use Enqueue\Psr\PsrMessage;
 
 class Promise
 {
     /**
-     * @var PsrConsumer
+     * @var \Closure
      */
-    private $consumer;
+    private $receiveCallback;
 
     /**
-     * @var int
+     * @var \Closure
      */
-    private $timeout;
+    private $receiveNoWaitCallback;
 
     /**
-     * @var string
+     * @var \Closure
      */
-    private $correlationId;
+    private $finallyCallback;
 
     /**
-     * @param PsrConsumer $consumer
-     * @param string      $correlationId
-     * @param int         $timeout
+     * @var bool
      */
-    public function __construct(PsrConsumer $consumer, $correlationId, $timeout)
+    private $deleteReplyQueue;
+
+    /**
+     * @var PsrMessage
+     */
+    private $message;
+
+    /**
+     * @param \Closure $receiveCallback
+     * @param \Closure $receiveNoWaitCallback
+     * @param \Closure $finallyCallback
+     */
+    public function __construct(\Closure $receiveCallback, \Closure $receiveNoWaitCallback, \Closure $finallyCallback)
     {
-        $this->consumer = $consumer;
-        $this->timeout = $timeout;
-        $this->correlationId = $correlationId;
+        $this->receiveCallback = $receiveCallback;
+        $this->receiveNoWaitCallback = $receiveNoWaitCallback;
+        $this->finallyCallback = $finallyCallback;
+
+        $this->deleteReplyQueue = true;
     }
 
     /**
+     * Blocks until message received or timeout expired.
+     *
+     * @deprecated use "receive" instead
+     *
      * @throws TimeoutException if the wait timeout is reached
      *
      * @return PsrMessage
      */
     public function getMessage()
     {
-        $endTime = time() + $this->timeout;
-
-        while (time() < $endTime) {
-            if ($message = $this->consumer->receive($this->timeout)) {
-                if ($message->getCorrelationId() === $this->correlationId) {
-                    $this->consumer->acknowledge($message);
+        return $this->receive();
+    }
 
-                    return $message;
+    /**
+     * Blocks until message received or timeout expired.
+     *
+     * @throws TimeoutException if the wait timeout is reached
+     *
+     * @return PsrMessage
+     */
+    public function receive()
+    {
+        if (null == $this->message) {
+            try {
+                if ($message = $this->doReceive($this->receiveCallback)) {
+                    $this->message = $message;
                 }
+            } finally {
+                call_user_func($this->finallyCallback, $this);
+            }
+        }
+
+        return $this->message;
+    }
+
+    /**
+     * Non blocking function. Returns message or null.
+     *
+     * @return PsrMessage|null
+     */
+    public function receiveNoWait()
+    {
+        if (null == $this->message) {
+            if ($message = $this->doReceive($this->receiveNoWaitCallback)) {
+                $this->message = $message;
 
-                $this->consumer->reject($message, true);
+                call_user_func($this->finallyCallback, $this);
             }
         }
 
-        throw TimeoutException::create($this->timeout, $this->correlationId);
+        return $this->message;
     }
 
     /**
-     * @param int $timeout
+     * On TRUE deletes reply queue after getMessage call.
+     *
+     * @param bool $delete
      */
-    public function setTimeout($timeout)
+    public function setDeleteReplyQueue($delete)
     {
-        $this->timeout = $timeout;
+        $this->deleteReplyQueue = (bool) $delete;
     }
 
     /**
-     * @return int
+     * @return bool
      */
-    public function getTimeout()
+    public function isDeleteReplyQueue()
     {
-        return $this->timeout;
+        return $this->deleteReplyQueue;
+    }
+
+    /**
+     * @param \Closure $cb
+     *
+     * @return PsrMessage
+     */
+    private function doReceive(\Closure $cb)
+    {
+        $message = call_user_func($cb, $this);
+
+        if (null !== $message && false == $message instanceof PsrMessage) {
+            throw new \RuntimeException(sprintf(
+                'Expected "%s" but got: "%s"', PsrMessage::class, is_object($message) ? get_class($message) : gettype($message)));
+        }
+
+        return $message;
     }
 }

pkg/enqueue/Rpc/RpcClient.php

@@ -33,7 +33,7 @@ public function __construct(PsrContext $context)
      */
     public function call(PsrDestination $destination, PsrMessage $message, $timeout)
     {
-        return $this->callAsync($destination, $message, $timeout)->getMessage();
+        return $this->callAsync($destination, $message, $timeout)->receive();
     }
 
     /**
@@ -51,9 +51,11 @@ public function callAsync(PsrDestination $destination, PsrMessage $message, $tim
 
         if ($message->getReplyTo()) {
             $replyQueue = $this->context->createQueue($message->getReplyTo());
+            $deleteReplyQueue = false;
         } else {
             $replyQueue = $this->context->createTemporaryQueue();
             $message->setReplyTo($replyQueue->getQueueName());
+            $deleteReplyQueue = true;
         }
 
         if (false == $message->getCorrelationId()) {
@@ -62,10 +64,54 @@ public function callAsync(PsrDestination $destination, PsrMessage $message, $tim
 
         $this->context->createProducer()->send($destination, $message);
 
-        return new Promise(
-            $this->context->createConsumer($replyQueue),
-            $message->getCorrelationId(),
-            $timeout
-        );
+        $correlationId = $message->getCorrelationId();
+
+        $receive = function () use ($replyQueue, $timeout, $correlationId) {
+            $endTime = time() + ((int) ($timeout / 1000));
+            $consumer = $this->context->createConsumer($replyQueue);
+
+            do {
+                if ($message = $consumer->receive($timeout)) {
+                    if ($message->getCorrelationId() === $correlationId) {
+                        $consumer->acknowledge($message);
+
+                        return $message;
+                    }
+
+                    $consumer->reject($message, true);
+                }
+            } while (time() < $endTime);
+
+            throw TimeoutException::create($timeout, $correlationId);
+        };
+
+        $receiveNoWait = function () use ($replyQueue, $correlationId) {
+            static $consumer;
+
+            if (null === $consumer) {
+                $consumer = $this->context->createConsumer($replyQueue);
+            }
+
+            if ($message = $consumer->receiveNoWait()) {
+                if ($message->getCorrelationId() === $correlationId) {
+                    $consumer->acknowledge($message);
+
+                    return $message;
+                }
+
+                $consumer->reject($message, true);
+            }
+        };
+
+        $finally = function (Promise $promise) use ($replyQueue) {
+            if ($promise->isDeleteReplyQueue() && method_exists($this->context, 'deleteQueue')) {
+                $this->context->deleteQueue($replyQueue);
+            }
+        };
+
+        $promise = new Promise($receive, $receiveNoWait, $finally);
+        $promise->setDeleteReplyQueue($deleteReplyQueue);
+
+        return $promise;
     }
 }