ProducerV2 For SimpleClient

pkg/enqueue/Client/ProducerV2.php

@@ -49,7 +49,7 @@ public function sendCommand($command, $message, $needReply = false)
         $message->setScope(Message::SCOPE_APP);
 
         if ($needReply) {
-            return $this->rpcClient->callAsync(Config::COMMAND_TOPIC, $message, 60);
+            return $this->rpcClient->callAsync(Config::COMMAND_TOPIC, $message, 60000);
         }
 
         $this->realProducer->send(Config::COMMAND_TOPIC, $message);

pkg/enqueue/Client/RpcClient.php

@@ -78,12 +78,13 @@ public function callAsync($topic, $message, $timeout)
 
         $correlationId = $message->getCorrelationId();
 
-        $receive = function () use ($replyQueue, $timeout, $correlationId) {
-            $endTime = time() + ((int) ($timeout / 1000));
+        $receive = function (Promise $promise, $promiseTimeout) use ($replyQueue, $timeout, $correlationId) {
+            $runTimeout = $promiseTimeout ?: $timeout;
+            $endTime = time() + ((int) ($runTimeout / 1000));
             $consumer = $this->context->createConsumer($replyQueue);
 
             do {
-                if ($message = $consumer->receive($timeout)) {
+                if ($message = $consumer->receive($runTimeout)) {
                     if ($message->getCorrelationId() === $correlationId) {
                         $consumer->acknowledge($message);
 
@@ -94,7 +95,7 @@ public function callAsync($topic, $message, $timeout)
                 }
             } while (time() < $endTime);
 
-            throw TimeoutException::create($timeout, $correlationId);
+            throw TimeoutException::create($runTimeout, $correlationId);
         };
 
         $receiveNoWait = function () use ($replyQueue, $correlationId) {

pkg/enqueue/Rpc/Promise.php

@@ -62,15 +62,17 @@ public function getMessage()
     /**
      * Blocks until message received or timeout expired.
      *
+     * @param int $timeout
+     *
      * @throws TimeoutException if the wait timeout is reached
      *
      * @return PsrMessage
      */
-    public function receive()
+    public function receive($timeout = null)
     {
         if (null == $this->message) {
             try {
-                if ($message = $this->doReceive($this->receiveCallback)) {
+                if ($message = $this->doReceive($this->receiveCallback, $this, $timeout)) {
                     $this->message = $message;
                 }
             } finally {
@@ -89,7 +91,7 @@ public function receive()
     public function receiveNoWait()
     {
         if (null == $this->message) {
-            if ($message = $this->doReceive($this->receiveNoWaitCallback)) {
+            if ($message = $this->doReceive($this->receiveNoWaitCallback, $this)) {
                 $this->message = $message;
 
                 call_user_func($this->finallyCallback, $this);
@@ -119,12 +121,13 @@ public function isDeleteReplyQueue()
 
     /**
      * @param \Closure $cb
+     * @param array    $args
      *
      * @return PsrMessage
      */
-    private function doReceive(\Closure $cb)
+    private function doReceive(\Closure $cb, ...$args)
     {
-        $message = call_user_func($cb, $this);
+        $message = call_user_func_array($cb, $args);
 
         if (null !== $message && false == $message instanceof PsrMessage) {
             throw new \RuntimeException(sprintf(

pkg/enqueue/Rpc/RpcClient.php

@@ -66,12 +66,13 @@ public function callAsync(PsrDestination $destination, PsrMessage $message, $tim
 
         $correlationId = $message->getCorrelationId();
 
-        $receive = function () use ($replyQueue, $timeout, $correlationId) {
-            $endTime = time() + ((int) ($timeout / 1000));
+        $receive = function (Promise $promise, $promiseTimeout) use ($replyQueue, $timeout, $correlationId) {
+            $runTimeout = $promiseTimeout ?: $timeout;
+            $endTime = time() + ((int) ($runTimeout / 1000));
             $consumer = $this->context->createConsumer($replyQueue);
 
             do {
-                if ($message = $consumer->receive($timeout)) {
+                if ($message = $consumer->receive($runTimeout)) {
                     if ($message->getCorrelationId() === $correlationId) {
                         $consumer->acknowledge($message);
 
@@ -82,7 +83,7 @@ public function callAsync(PsrDestination $destination, PsrMessage $message, $tim
                 }
             } while (time() < $endTime);
 
-            throw TimeoutException::create($timeout, $correlationId);
+            throw TimeoutException::create($runTimeout, $correlationId);
         };
 
         $receiveNoWait = function () use ($replyQueue, $correlationId) {

pkg/enqueue/Tests/Client/RpcClientTest.php

@@ -174,6 +174,7 @@ public function testShouldReceiveMessageAndAckMessageIfCorrelationEquals()
         $consumer
             ->expects($this->once())
             ->method('receive')
+            ->with(12345)
             ->willReturn($receivedMessage)
         ;
         $consumer
@@ -202,7 +203,7 @@ public function testShouldReceiveMessageAndAckMessageIfCorrelationEquals()
 
         $rpc = new RpcClient($this->createProducerMock(), $context);
 
-        $rpc->callAsync('topic', $message, 2)->receive();
+        $rpc->callAsync('topic', $message, 2)->receive(12345);
     }
 
     public function testShouldReceiveNoWaitMessageAndAckMessageIfCorrelationEquals()

pkg/enqueue/Tests/Rpc/PromiseTest.php

@@ -29,14 +29,39 @@ public function testCouldSetGetDeleteReplyQueue()
     public function testOnReceiveShouldCallReceiveCallBack()
     {
         $receiveInvoked = false;
-        $receivecb = function () use (&$receiveInvoked) {
+        $receivePromise = null;
+        $receiveTimeout = null;
+        $receivecb = function ($promise, $timout) use (&$receiveInvoked, &$receivePromise, &$receiveTimeout) {
             $receiveInvoked = true;
+            $receivePromise = $promise;
+            $receiveTimeout = $timout;
         };
 
         $promise = new Promise($receivecb, function () {}, function () {});
         $promise->receive();
 
         $this->assertTrue($receiveInvoked);
+        $this->assertInstanceOf(Promise::class, $receivePromise);
+        $this->assertNull($receiveTimeout);
+    }
+
+    public function testOnReceiveShouldCallReceiveCallBackWithTimeout()
+    {
+        $receiveInvoked = false;
+        $receivePromise = null;
+        $receiveTimeout = null;
+        $receivecb = function ($promise, $timout) use (&$receiveInvoked, &$receivePromise, &$receiveTimeout) {
+            $receiveInvoked = true;
+            $receivePromise = $promise;
+            $receiveTimeout = $timout;
+        };
+
+        $promise = new Promise($receivecb, function () {}, function () {});
+        $promise->receive(12345);
+
+        $this->assertTrue($receiveInvoked);
+        $this->assertInstanceOf(Promise::class, $receivePromise);
+        $this->assertSame(12345, $receiveTimeout);
     }
 
     public function testOnReceiveNoWaitShouldCallReceiveNoWaitCallBack()

pkg/enqueue/Tests/Rpc/RpcClientTest.php

@@ -114,6 +114,7 @@ public function testShouldReceiveMessageAndAckMessageIfCorrelationEquals()
         $consumer
             ->expects($this->once())
             ->method('receive')
+            ->with(12345)
             ->willReturn($receivedMessage)
         ;
         $consumer
@@ -147,7 +148,7 @@ public function testShouldReceiveMessageAndAckMessageIfCorrelationEquals()
 
         $rpc = new RpcClient($context);
 
-        $rpc->callAsync($queue, $message, 2)->receive();
+        $rpc->callAsync($queue, $message, 2)->receive(12345);
     }
 
     public function testShouldReceiveNoWaitMessageAndAckMessageIfCorrelationEquals()

pkg/simple-client/SimpleClient.php

@@ -11,6 +11,7 @@
 use Enqueue\Client\Meta\QueueMetaRegistry;
 use Enqueue\Client\Meta\TopicMetaRegistry;
 use Enqueue\Client\ProducerInterface;
+use Enqueue\Client\ProducerV2Interface;
 use Enqueue\Client\RouterProcessor;
 use Enqueue\Consumption\CallbackProcessor;
 use Enqueue\Consumption\ExtensionInterface;
@@ -185,6 +186,18 @@ public function getProducer($setupBroker = false)
         return $this->container->get('enqueue.client.producer');
     }
 
+    /**
+     * @param bool $setupBroker
+     *
+     * @return ProducerV2Interface
+     */
+    public function getProducerV2($setupBroker = false)
+    {
+        $setupBroker && $this->setupBroker();
+
+        return $this->container->get('enqueue.client.producer.v2');
+    }
+
     public function setupBroker()
     {
         $this->getDriver()->setupBroker();

pkg/simple-client/SimpleClientContainerExtension.php

@@ -10,7 +10,9 @@
 use Enqueue\Client\Meta\QueueMetaRegistry;
 use Enqueue\Client\Meta\TopicMetaRegistry;
 use Enqueue\Client\Producer;
+use Enqueue\Client\ProducerV2;
 use Enqueue\Client\RouterProcessor;
+use Enqueue\Client\RpcClient;
 use Enqueue\Consumption\ChainExtension as ConsumptionChainExtension;
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\Symfony\TransportFactoryInterface;
@@ -93,6 +95,18 @@ public function load(array $configs, ContainerBuilder $container)
                 new Reference('enqueue.client.driver'),
         ]);
 
+        $container->register('enqueue.client.rpc', RpcClient::class)
+            ->setArguments([
+                new Reference('enqueue.client.producer'),
+                new Reference('enqueue.transport.context'),
+        ]);
+
+        $container->register('enqueue.client.producer.v2', ProducerV2::class)
+            ->setArguments([
+                new Reference('enqueue.client.producer'),
+                new Reference('enqueue.client.rpc'),
+        ]);
+
         $container->register('enqueue.client.meta.topic_meta_registry', TopicMetaRegistry::class)
             ->setArguments([[]]);