Merge pull request #25 from skysbird/php-amqplib

add custom routing_key and expire time for CELERY_TASK_RESULT_EXPIRES

amqplibconnector.php

@@ -1,7 +1,6 @@
 <?php
 
 require_once('amqp.php');
-require_once('vendor/autoload.php');
 
 use PhpAmqpLib\Connection\AMQPConnection;
 use PhpAmqpLib\Message\AMQPMessage;
@@ -76,7 +75,7 @@ function PostToExchange($connection, $details, $task, $params)
 			$params
 		);
 
-		$ch->basic_publish($msg, $details['exchange']);
+		$ch->basic_publish($msg, $details['exchange'],$details['routing_key']);
 
 		$ch->close();
 	}
@@ -94,21 +93,28 @@ function Consume($msg)
 	 * Return result of task execution for $task_id
 	 * @param object $connection AMQPConnection object
 	 * @param string $task_id Celery task identifier
+	 * @param int $expire expire time result queue, milliseconds
 	 * @return array array('body' => JSON-encoded message body, 'complete_result' => AMQPMessage object)
 	 * 			or false if result not ready yet
 	 */
-	function GetMessageBody($connection, $task_id)
+	function GetMessageBody($connection, $task_id,$expire=0)
 	{
 		if(!$this->receiving_channel)
 		{
 			$ch = $connection->channel();
+            $expire_args = null;
+            if(!empty($expire)){
+                $expire_args = array("x-expires"=>array("I",$expire));
+            }
 
 			$ch->queue_declare(
 				$task_id, 				/* queue name */
 				false,					/* passive */
-				false,					/* durable */
+				true,					/* durable */
 				false,					/* exclusive */
-				true					/* auto_delete */
+				true,					/* auto_delete */
+                false,                  /*no wait*/
+                $expire_args
 			);
 
 			$ch->queue_bind($task_id, 'celeryresults');

celery.php

@@ -64,9 +64,9 @@ class Celery
 	private $connection_details = array(); // array of strings required to connect
 	private $amqp = null; // AbstractAMQPConnector implementation
 
-	function __construct($host, $login, $password, $vhost, $exchange='celery', $binding='celery', $port=5672, $connector=false, $persistent_messages=false)
+	function __construct($host, $login, $password, $vhost, $exchange='celery', $binding='celery', $port=5672, $connector=false, $persistent_messages=false,$result_expire=0)
 	{
-		foreach(array('host', 'login', 'password', 'vhost', 'exchange', 'binding', 'port', 'connector', 'persistent_messages') as $detail)
+		foreach(array('host', 'login', 'password', 'vhost', 'exchange', 'binding', 'port', 'connector', 'persistent_messages','result_expire') as $detail)
 		{
 			$this->connection_details[$detail] = $$detail;
 		}
@@ -94,7 +94,7 @@ static function InitializeAMQPConnection($details)
 	 * @param array $args Array of arguments (kwargs call when $args is associative)
 	 * @return AsyncResult
 	 */
-	function PostTask($task, $args, $async_result=true)
+	function PostTask($task, $args, $async_result=true,$routing_key="celery")
 	{
 		if(!is_array($args))
 		{
@@ -131,6 +131,8 @@ function PostTask($task, $args, $async_result=true)
 			$params['delivery_mode'] = 2;
 		}
 
+        $this->connection_details['routing_key'] = $routing_key;
+
 		$success = $this->amqp->PostToExchange(
 			$this->connection,
 			$this->connection_details,
@@ -195,7 +197,7 @@ private function getCompleteResult()
 			return $this->complete_result;
 		}
 
-		$message = $this->amqp->GetMessageBody($this->connection, $this->task_id);
+		$message = $this->amqp->GetMessageBody($this->connection, $this->task_id,$this->connection_details['result_expire']);
 
 		if($message !== false)
 		{