repo_mongoAdapter/lib/Mooses/Mongodb/Queue/Mongoadapter.php

<?php
require_once 'Zend/Queue/Adapter/AdapterAbstract.php';

/**
 * Class for using a standard PHP array as a queue
 *
 * @category   Zend
 * @package    Zend_Queue
 * @subpackage Adapter
 * @copyright  Copyright (c) 2005-2012 Zend Technologies USA Inc. (http://www.zend.com)
 * @license    http://framework.zend.com/license/new-bsd     New BSD License
 */
class Mooses_Mongodb_Queue_Mongoadapter extends Zend_Queue_Adapter_AdapterAbstract {
    /**
     * @var array
     */
    protected $_data = array();
    protected $_dataContainer = NULL;
    protected $_queueCollection = NULL;
    protected static $instance = NULL;

    /**
     * Constructor
     *
     * @param  array|Zend_Config $options
     * @param  Zend_Queue|null $queue
     * @return void
     */
    public function __construct($options, Zend_Queue $queue = null)
    {
        $this->_dataContainer = new Mooses_Mongodb_Queue_Abstractadapter();
        $this->_queueCollection = new Mooses_Mongodb_Queue_Queueadapter();
        if(is_null($options)){
            $options = array();
        }
        parent::__construct($options, $queue);
    }

    public static function getInstance() {
        if(is_null(self::$instance)) {
            $_calledClass = get_called_class();
            self::$instance = new $_calledClass();
        }
        return self::$instance;
    }

    /********************************************************************
     * Queue management functions
     *********************************************************************/

    /**
     * Does a queue already exist?
     *
     * Throws an exception if the adapter cannot determine if a queue exists.
     * use isSupported('isExists') to determine if an adapter can test for
     * queue existance.
     *
     * @param  string $name
     * @return boolean
     */
    public function isExists($name)
    {
        try {
            $_existantQueue = $this->_queueCollection->loadByAttribute("queue_name", $name);
            return (($_existantQueue) ? true : false);
        } catch(Exception $e){
            var_dump($e->getMessage());
            die($e->getTraceAsString());
        }
    }

    /**
     * Create a new queue
     *
     * Visibility timeout is how long a message is left in the queue "invisible"
     * to other readers.  If the message is acknowleged (deleted) before the
     * timeout, then the message is deleted.  However, if the timeout expires
     * then the message will be made available to other queue readers.
     *
     * @param  string  $name    queue name
     * @param  integer $timeout default visibility timeout
     * @return boolean
     */
    public function create($name, $timeout=null)
    {
        if ($this->isExists($name)) {
            return false;
        }
        if ($timeout === null) {
            $timeout = self::CREATE_TIMEOUT_DEFAULT;
        }
        $_newQueue = $this->_queueCollection->setData("queue_id", uniqid("vfq_"))
            ->setData("queue_name", $name)
            ->setData("timeout", 86400);
        $_newQueue->save(true);
        return true;
    }

    /**
     * Delete a queue and all of it's messages
     *
     * Returns false if the queue is not found, true if the queue exists
     *
     * @param  string  $name queue name
     * @return boolean
     */
    public function delete($name)
    {
        try {
            $_existantQueue = $this->_queueCollection->loadByAttribute("name", $name);
            if ($_existantQueue) {
                $_collectionMessages = $this->_dataContainer->getCollection()->addAttributeToFilter("queue_id", $_existantQueue->getData("queue_id"))->getCollectionData();
            foreach ($_collectionMessages as $_message) {
                $_message->delete(false);
            }
            $this->_dataContainer->load($_existantQueue->getId())->delete(false);
        }
            return true;
        } catch (Exception $e){
            return false;
        }
    }

    /**
     * Get an array of all available queues
     *
     * Not all adapters support getQueues(), use isSupported('getQueues')
     * to determine if the adapter supports this feature.
     *
     * @return array
     */
    public function getQueues()
    {
        $_collectionQueues = $this->_queueCollection->getCollection()->getCollectionData();
        return $_collectionQueues;
    }

    /**
     * Return the approximate number of messages in the queue
     *
     * @param  Zend_Queue $queue
     * @return integer
     * @throws Zend_Queue_Exception
     */
    public function count(Zend_Queue $queue=null)
    {
        if ($queue === null) {
            $queue = $this->queue;
        }

        if ($this->isExists($queue->getName())) {
            /**
             * @see Zend_Queue_Exception
             */
//            require_once 'Zend/Queue/Exception.php';
            throw new Zend_Queue_Exception('Queue does not exist');
        }
        $_messages = $queue->count();
        return $_messages;
    }

    /********************************************************************
     * Messsage management functions
     *********************************************************************/

    /**
     * Send a message to the queue
     *
     * @param  string     $message Message to send to the active queue
     * @param  Zend_Queue $queue
     * @return Zend_Queue_Message
     * @throws Zend_Queue_Exception
     */
    public function send($message, Zend_Queue $queue=null)
    {
        if ($queue === null) {
            $queue = $this->_queue;
        }

        if (!$this->isExists($queue->getName())) {
            throw new Zend_Queue_Exception('Queue does not exist:' . $queue->getName());
        }

        // create the message
        $data = array(
            'message_id' => md5(uniqid(rand(), true)),
            'body'       => $message,
            'md5'        => md5($message),
            'handle'     => null,
            'created'    => time(),
            'queue_name' => $queue->getName(),
            'retrials'   => 0
        );

        // add $data to the "queue"
        $_queueAdapter = $queue->getAdapter();
        $_queueAdapter->setData($data)->save();

        $options = array(
            'queue' => $queue,
            'data'  => $data,
        );

        $classname = "Mooses_Mongodb_Queue_Abstractadapter";
        if (!class_exists($classname)) {
            require_once 'Zend/Loader.php';
            Zend_Loader::loadClass($classname);
        }
        return new $classname($options);
    }

    /**
     * Get messages in the queue
     *
     * @param  integer    $maxMessages  Maximum number of messages to return
     * @param  integer    $timeout      Visibility timeout for these messages
     * @param  Zend_Queue $queue
     * @return Zend_Queue_Message_Iterator
     */
    public function receive($maxMessages = null, $timeout = 1800, Zend_Queue $queue = null)
    {
        if ($maxMessages === null) {
            $maxMessages = 1;
        }
        if ($timeout === null) {
            $timeout = self::RECEIVE_TIMEOUT_DEFAULT;
        }
        if ($queue === null) {
            $queue = $this->_queue;
        }
        $data = array();
        if ($maxMessages > 0) {
            $start_time = microtime(true);

            $count = 0;
            $_collectionModel = new Mooses_Mongodb_Queue_Abstractadapter();
            $_collectionData = $_collectionModel
                ->loadByAttribute("queue_name", $queue->getName(), true);
            foreach(array_reverse($_collectionData) as $key => $msg){
                if ($count >= $maxMessages) {
                    break;
                }
                $_messageId = $msg->getData("message_id");
                $_retrials = (int) $msg->getData("retrials");
                if (($msg->getData('handle') === null) || (!is_null($msg->getData("handle")) && $_retrials < 3)) {
                    $_randomId = md5(uniqid(rand(), true));
                    $_microtime = microtime(true);
                    $_retrials = (int) ($_retrials + 1);
                    $msg->setData('handle', $_randomId)->setData("timeout", $_microtime);
                    $msg->update(array("message_id" => $_messageId), array('$set' => array("handle" => $_randomId, "timeout" => $_microtime, "retrials" => $_retrials)));
                    $data[] = $msg->getData();
                    $count++;
//                } elseif(($msg->getData('handle') === null && $_timeout < $start_time)){
//                    $_logger = Mooses_Logger::getInstance("vfctc_ordersynch_error");
//                    $_logger->log("SYNCH_ORDERS", Zend_Log::WARN, "Too early to process this message: {\"message_id\": \"" . $_messageId . "\"}");
                } elseif($_retrials > 2){
                    $_logger = Mooses_Logger::getInstance("vfctc_ordersynch_error");
                    $_logger->log("SYNCH_ORDERS", Zend_Log::EMERG, "Too many retrials on message: {\"message_id\": \"" . $_messageId . "\"}");
                    $msg->update(array("message_id" => $_messageId), array('$set' => array("queue_name" => "error_queue")));
                }

            }
        }

        $options = array(
            'queue'        => $queue,
            'data'         => $data,
            'messageClass' => $queue->getMessageClass(),
        );
        $classname = $queue->getMessageSetClass();
        if (!class_exists($classname)) {
            require_once 'Zend/Loader.php';
            Zend_Loader::loadClass($classname);
        }
        return new $classname($options);
    }

    public function addRetryErrorMessage($_messageId, $_error, $_oldRetrialMessages = array()){
        $_collectionModel = new Mooses_Mongodb_Queue_Abstractadapter();
        $_message = $_collectionModel->loadByAttribute("message_id", $_messageId, false);
        if($_message){
            $_retrialMessages = array();
            if(is_array($_oldRetrialMessages)) {
                $_retrialMessages = array_merge($_retrialMessages, $_oldRetrialMessages);
            }
            if(is_object($_error) || is_array($_error)){
                $_error = json_encode($_error);
            }
            array_push($_retrialMessages, array("date" => date("Y-m-d H:i:s"), "message" => $_error));
            $_collectionModel->update(array("message_id" => $_messageId), array('$set' => array("retrial_messages" => $_retrialMessages)));
        };
    }

    /**
     * Delete a message from the queue
     *
     * Returns true if the message is deleted, false if the deletion is
     * unsuccessful.
     *
     * @param  Zend_Queue_Message $message
     * @return boolean
     * @throws Zend_Queue_Exception
     */
    public function deleteMessage(Zend_Queue_Message $message)
    {
        try {
            $queue = new Mooses_Mongodb_Queue_Abstractadapter();
            $_messageData = $message->toArray();
            return $queue->remove(array("handle" => $_messageData['handle']), array("justOne" => true));
        } catch (Exception $e){
            return false;
        }
    }

    /********************************************************************
     * Supporting functions
     *********************************************************************/

    /**
     * Return a list of queue capabilities functions
     *
     * $array['function name'] = true or false
     * true is supported, false is not supported.
     *
     * @param  string $name
     * @return array
     */
    public function getCapabilities()
    {
        return array(
            'create'        => true,
            'delete'        => true,
            'send'          => true,
            'receive'       => true,
            'deleteMessage' => true,
            'getQueues'     => true,
            'count'         => true,
            'isExists'      => true,
        );
    }

    /********************************************************************
     * Functions that are not part of the Zend_Queue_Adapter_Abstract
     *********************************************************************/

    /**
     * serialize
     */
    public function __sleep()
    {
        return array('_data');
    }

    /*
     * These functions are debug helpers.
     */

    /**
     * returns underlying _data array
     * $queue->getAdapter()->getData();
     *
     * @return $this;
     */
    public function getData()
    {
        //$this->_dataContainer->
    }

    /**
     * sets the underlying _data array
     * $queue->getAdapter()->setData($data);
     *
     * @param array $data
     * @return $this;
     */
    public function setData($data)
    {
        if(is_array($data)) {
            foreach ($data as $_key => $_singleData) {
                $this->_dataContainer->setData($_key, $_singleData);
            }
        }
        return $this;
    }

    public function save(){
        return $this->_dataContainer->save(true, false);
    }

}