Added possibility to set serialize/unserialize function for rpc servers/clients

Author: igaponov

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

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);
         }
     }

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;
+    }
 }

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;
+    }
 }