Merge branch 'mvonahn-master'

Author: John Kelly

lib/Thumper/BaseAmqp.php

@@ -39,6 +39,9 @@
 
 abstract class BaseAmqp
 {
+    const NON_PERSISTENT = 1;
+    const PERSISTENT = 2;
+    
     /**
      * @var AbstractConnection
      */
@@ -88,6 +91,13 @@ abstract class BaseAmqp
      */
     protected $routingKey = '';
 
+    /**
+     * @var array
+     */
+    protected $parameters = array(
+        'content_type' => 'text/plain'
+    );
+
     /**
      * BaseAmqp constructor.
      * @param AbstractConnection $connection
@@ -226,4 +236,21 @@ private function isValidExchangeName($exchangeName)
     {
         return preg_match('/^[A-Za-z0-9_\-\.\;]*$/', $exchangeName);
     }
+
+    /**
+     * @param string $key
+     * @param string $value
+     */
+    public function setParameter($key, $value)
+    {
+        $this->parameters[$key] = $value;
+    }
+
+    /**
+     * @return array
+     */
+    public function getParameters()
+    {
+        return $this->parameters;
+    }
 }

lib/Thumper/Producer.php

@@ -61,10 +61,12 @@ public function publish($messageBody, $routingKey = '')
             }
             $this->exchangeReady = true;
         }
+        
+        $this->setParameter('delivery_mode', self::PERSISTENT);
 
         $message = new AMQPMessage(
             $messageBody,
-            array('content_type' => 'text/plain', 'delivery_mode' => 2)
+            $this->getParameters()
         );
         $this->channel->basic_publish(
             $message,

lib/Thumper/RpcClient.php

@@ -80,14 +80,12 @@ public function addRequest($messageBody, $server, $requestId, $routingKey = '')
         if (empty($requestId)) {
             throw new \InvalidArgumentException("You must provide a request ID");
         }
+        $this->setParameter('correlation_id', $requestId);
+        $this->setParameter('reply_to', $this->queueName);
 
         $message = new AMQPMessage(
             $messageBody,
-            array(
-                'content_type' => 'text/plain',
-                'reply_to' => $this->queueName,
-                'correlation_id' => $requestId
-            )
+            $this->getParameters()
         );
 
         $this->channel

lib/Thumper/RpcServer.php

@@ -103,12 +103,10 @@ public function processMessage(AMQPMessage $message)
      */
     protected function sendReply($result, $client, $correlationId)
     {
+        $this->setParameter('correlation_id', $correlationId);
         $reply = new AMQPMessage(
             $result,
-            array(
-                'content_type' => 'text/plain',
-                'correlation_id' => $correlationId
-            )
+            $this->getParameters()
         );
         $this->channel
             ->basic_publish($reply, '', $client);