Merge pull request #238 from igaponov/custom-serializer

videlalvaro · videlalvaro · commit 3dd1b63f9853 · 2015-03-09T12:37:42.000+01:00
Add serialize/unserialize function for rpc
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,3 +1,6 @@
+- 2015-02-07
+    * Added possibility to set serialize/unserialize function for rpc servers/clients
+
 - 2014-11-27
     * Added interface `OldSound\RabbitMqBundle\Provider\QueuesProviderInterface`
     * Added `queues_provider` configuration for multiple consumer
diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
@@ -185,6 +185,7 @@ protected function addRpcClients(ArrayNodeDefinition $node)
                         ->children()
                             ->scalarNode('connection')->defaultValue('default')->end()
                             ->booleanNode('expect_serialized_response')->defaultTrue()->end()
+                            ->scalarNode('unserializer')->defaultValue('unserialize')->end()
                         ->end()
                     ->end()
                 ->end()
@@ -213,6 +214,7 @@ protected function addRpcServers(ArrayNodeDefinition $node)
                                     ->booleanNode('global')->defaultFalse()->end()
                                 ->end()
                             ->end()
+                            ->scalarNode('serializer')->defaultValue('serialize')->end()
                         ->end()
                     ->end()
                 ->end()
diff --git a/DependencyInjection/OldSoundRabbitMqExtension.php b/DependencyInjection/OldSoundRabbitMqExtension.php
@@ -312,6 +312,9 @@ protected function loadRpcClients()
             if ($this->collectorEnabled) {
                 $this->injectLoggedChannel($definition, $key, $client['connection']);
             }
+            if (array_key_exists('unserializer', $client)) {
+                $definition->addMethodCall('setUnserializer', array($client['unserializer']));
+            }
 
             $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_rpc', $key), $definition);
         }
@@ -340,6 +343,9 @@ protected function loadRpcServers()
             if (array_key_exists('queue_options', $server)) {
                 $definition->addMethodCall('setQueueOptions', array($server['queue_options']));
             }
+            if (array_key_exists('serializer', $server)) {
+                $definition->addMethodCall('setSerializer', array($server['serializer']));
+            }
             $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_server', $key), $definition);
         }
     }
diff --git a/README.md b/README.md
@@ -380,12 +380,14 @@ Let's add a RPC client and server into the configuration:
 rpc_clients:
     integer_store:
         connection: default
+        unserializer: json_decode
 rpc_servers:
     random_int:
         connection: default
         callback:   random_int_server
         qos_options: {prefetch_size: 0, prefetch_count: 1, global: false}
         queue_options: {name: random_int_queue, durable: false, auto_delete: true}
+        serializer: json_encode
 ```
 
 *For a full configuration reference please use the `php app/console config:dump-reference old_sound_rabbit_mq` command.*
diff --git a/RabbitMq/RpcClient.php b/RabbitMq/RpcClient.php
@@ -13,6 +13,7 @@ class RpcClient extends BaseAmqp
     protected $timeout = 0;
 
     private $queueName;
+    private $unserializer = 'unserialize';
 
     public function initClient($expectSerializedResponse = true)
     {
@@ -60,7 +61,7 @@ public function processMessage(AMQPMessage $msg)
     {
         $messageBody = $msg->body;
         if ($this->expectSerializedResponse) {
-            $messageBody = unserialize($messageBody);
+            $messageBody = call_user_func($this->unserializer, $messageBody);
         }
 
         $this->replies[$msg->get('correlation_id')] = $messageBody;
@@ -74,4 +75,9 @@ protected function getQueueName()
 
         return $this->queueName;
     }
+
+    public function setUnserializer($unserializer)
+    {
+        $this->unserializer = $unserializer;
+    }
 }
diff --git a/RabbitMq/RpcServer.php b/RabbitMq/RpcServer.php
@@ -6,6 +6,8 @@
 
 class RpcServer extends BaseConsumer
 {
+    private $serializer = 'serialize';
+
     public function initServer($name)
     {
         $this->setExchangeOptions(array('name' => $name, 'type' => 'direct'));
@@ -17,7 +19,8 @@ public function processMessage(AMQPMessage $msg)
         try {
             $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
             $result = call_user_func($this->callback, $msg);
-            $this->sendReply(serialize($result), $msg->get('reply_to'), $msg->get('correlation_id'));
+            $result = call_user_func($this->serializer, $result);
+            $this->sendReply($result, $msg->get('reply_to'), $msg->get('correlation_id'));
             $this->consumed++;
             $this->maybeStopConsumer();
         } catch (\Exception $e) {
@@ -30,4 +33,9 @@ protected function sendReply($result, $client, $correlationId)
         $reply = new AMQPMessage($result, array('content_type' => 'text/plain', 'correlation_id' => $correlationId));
         $this->getChannel()->basic_publish($reply, '', $client);
     }
+
+    public function setSerializer($serializer)
+    {
+        $this->serializer = $serializer;
+    }
 }
diff --git a/Tests/DependencyInjection/Fixtures/test.yml b/Tests/DependencyInjection/Fixtures/test.yml
@@ -147,13 +147,15 @@ old_sound_rabbit_mq:
     rpc_clients:
         foo_client:
             connection:      foo_connection
+            unserializer:    json_decode
 
         default_client:
 
     rpc_servers:
         foo_server:
             connection:      foo_connection
             callback:        foo_server.callback
+            serializer:      json_encode
 
         default_server:
             callback:        default_server.callback
diff --git a/Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php b/Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php
@@ -473,7 +473,10 @@ public function testFooRpcClientDefinition()
         $this->assertEquals((string) $definition->getArgument(0), 'old_sound_rabbit_mq.connection.foo_connection');
         $this->assertEquals((string) $definition->getArgument(1), 'old_sound_rabbit_mq.channel.foo_client');
         $this->assertEquals(
-            array(array('initClient', array(true))),
+            array(
+                array('initClient', array(true)),
+                array('setUnserializer', array('json_decode'))
+            ),
             $definition->getMethodCalls()
         );
         $this->assertEquals('%old_sound_rabbit_mq.rpc_client.class%', $definition->getClass());
@@ -488,7 +491,10 @@ public function testDefaultRpcClientDefinition()
         $this->assertEquals((string) $definition->getArgument(0), 'old_sound_rabbit_mq.connection.default');
         $this->assertEquals((string) $definition->getArgument(1), 'old_sound_rabbit_mq.channel.default_client');
         $this->assertEquals(
-            array(array('initClient', array(true))),
+            array(
+                array('initClient', array(true)),
+                array('setUnserializer', array('unserialize'))
+            ),
             $definition->getMethodCalls()
         );
         $this->assertEquals('%old_sound_rabbit_mq.rpc_client.class%', $definition->getClass());
@@ -505,6 +511,7 @@ public function testFooRpcServerDefinition()
         $this->assertEquals(array(
                 array('initServer', array('foo_server')),
                 array('setCallback', array(array(new Reference('foo_server.callback'), 'execute'))),
+                array('setSerializer', array('json_encode')),
             ),
             $definition->getMethodCalls()
         );
@@ -522,6 +529,7 @@ public function testDefaultRpcServerDefinition()
         $this->assertEquals(array(
                 array('initServer', array('default_server')),
                 array('setCallback', array(array(new Reference('default_server.callback'), 'execute'))),
+                array('setSerializer', array('serialize')),
             ),
             $definition->getMethodCalls()
         );
@@ -550,6 +558,7 @@ public function testRpcServerWithQueueOptionsDefinition()
                     'ticket'       => null,
                     'routing_keys' => array(),
                 ))),
+                array('setSerializer', array('serialize')),
             ),
             $definition->getMethodCalls()
         );
diff --git a/Tests/RabbitMq/RpcClientTest.php b/Tests/RabbitMq/RpcClientTest.php
@@ -0,0 +1,25 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Tests\RabbitMq;
+
+use OldSound\RabbitMqBundle\RabbitMq\RpcClient;
+
+class RpcClientTest extends \PHPUnit_Framework_TestCase
+{
+    public function testProcessMessageWithCustomUnserializer()
+    {
+        /** @var RpcClient $client */
+        $client = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\RpcClient')
+            ->setMethods(array('sendReply', 'maybeStopConsumer'))
+            ->disableOriginalConstructor()
+            ->getMock();
+        $message = $this->getMock('\PhpAmqpLib\Message\AMQPMessage', array('get'), array('message'));
+        $serializer = $this->getMock('\Symfony\Component\Serializer\SerializerInterface', array('serialize', 'deserialize'));
+        $serializer->expects($this->once())->method('deserialize')->with('message', 'json', null);
+        $client->initClient(true);
+        $client->setUnserializer(function($data) use ($serializer) {
+            $serializer->deserialize($data, 'json', null);
+        });
+        $client->processMessage($message);
+    }
+}
diff --git a/Tests/RabbitMq/RpcServerTest.php b/Tests/RabbitMq/RpcServerTest.php
@@ -0,0 +1,32 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Tests\RabbitMq;
+
+use OldSound\RabbitMqBundle\RabbitMq\RpcServer;
+use PhpAmqpLib\Message\AMQPMessage;
+
+class RpcServerTest extends \PHPUnit_Framework_TestCase
+{
+    public function testProcessMessageWithCustomSerializer()
+    {
+        /** @var RpcServer $server */
+        $server = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\RpcServer')
+            ->setMethods(array('sendReply', 'maybeStopConsumer'))
+            ->disableOriginalConstructor()
+            ->getMock();
+        $message = $this->getMock('\PhpAmqpLib\Message\AMQPMessage', array('get'));
+        $message->delivery_info = array(
+            'channel' => $this->getMock('\PhpAmqpLib\Channel\AMQPChannel', array(), array(), '', false),
+            'delivery_tag' => null
+        );
+        $server->setCallback(function() {
+            return 'message';
+        });
+        $serializer = $this->getMock('\Symfony\Component\Serializer\SerializerInterface', array('serialize', 'deserialize'));
+        $serializer->expects($this->once())->method('serialize')->with('message', 'json');
+        $server->setSerializer(function($data) use ($serializer) {
+            $serializer->serialize($data, 'json');
+        });
+        $server->processMessage($message);
+    }
+}
diff --git a/composer.json b/composer.json
@@ -14,7 +14,8 @@
         "videlalvaro/php-amqplib"  : "~2.4.0"
     },
     "require-dev": {
-        "symfony/console": "~2.0"
+        "symfony/console": "~2.0",
+        "symfony/serializer": "~2.6"
     },
     "extra": {
         "branch-alias": {

Original file line number	Diff line number	Diff line change
`@@ -312,6 +312,9 @@ protected function loadRpcClients()`
`312`	`312`	`if ($this->collectorEnabled) {`
`313`	`313`	`$this->injectLoggedChannel($definition, $key, $client['connection']);`
`314`	`314`	`}`
	`315`	`+ if (array_key_exists('unserializer', $client)) {`
	`316`	`+ $definition->addMethodCall('setUnserializer', array($client['unserializer']));`
	`317`	`+ }`
`315`	`318`
`316`	`319`	`$this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_rpc', $key), $definition);`
`317`	`320`	`}`
`@@ -340,6 +343,9 @@ protected function loadRpcServers()`
`340`	`343`	`if (array_key_exists('queue_options', $server)) {`
`341`	`344`	`$definition->addMethodCall('setQueueOptions', array($server['queue_options']));`
`342`	`345`	`}`
	`346`	`+ if (array_key_exists('serializer', $server)) {`
	`347`	`+ $definition->addMethodCall('setSerializer', array($server['serializer']));`
	`348`	`+ }`
`343`	`349`	`$this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_server', $key), $definition);`
`344`	`350`	`}`
`345`	`351`	`}`
Original file line number	Diff line number	Diff line change
`@@ -13,6 +13,7 @@ class RpcClient extends BaseAmqp`
`13`	`13`	`protected $timeout = 0;`
`14`	`14`
`15`	`15`	`private $queueName;`
	`16`	`+ private $unserializer = 'unserialize';`
`16`	`17`
`17`	`18`	`public function initClient($expectSerializedResponse = true)`
`18`	`19`	`{`
`@@ -60,7 +61,7 @@ public function processMessage(AMQPMessage $msg)`
`60`	`61`	`{`
`61`	`62`	`$messageBody = $msg->body;`
`62`	`63`	`if ($this->expectSerializedResponse) {`
`63`		`- $messageBody = unserialize($messageBody);`
	`64`	`+ $messageBody = call_user_func($this->unserializer, $messageBody);`
`64`	`65`	`}`
`65`	`66`
`66`	`67`	`$this->replies[$msg->get('correlation_id')] = $messageBody;`
`@@ -74,4 +75,9 @@ protected function getQueueName()`
`74`	`75`
`75`	`76`	`return $this->queueName;`
`76`	`77`	`}`
	`78`	`+`
	`79`	`+ public function setUnserializer($unserializer)`
	`80`	`+ {`
	`81`	`+ $this->unserializer = $unserializer;`
	`82`	`+ }`
`77`	`83`	`}`