sintonia/airtime_mvc/application/services/ThirdPartyCeleryService.php

<?php

abstract class Application_Service_ThirdPartyCeleryService extends Application_Service_ThirdPartyService {

    /**
     * @var string broker exchange name for third-party tasks
     */
    protected static $_CELERY_EXCHANGE_NAME;

    /**
     * @var array map of celery identifiers to their task names
     */
    protected static $_CELERY_TASKS;

    /**
     * Execute a Celery task with the given name and data parameters
     *
     * @param string $taskName the name of the celery task to execute
     * @param array $data      the data array to send as task parameters
     * @param int $fileId      the unique identifier for the file involved in the task
     *
     * @return CeleryTasks the created task
     *
     * @throws Exception
     */
    protected function _executeTask($taskName, $data, $fileId = null) {
        try {
            $brokerTaskId = CeleryManager::sendCeleryMessage($taskName,
                                                             static::$_CELERY_EXCHANGE_NAME,
                                                             $data);
            return $this->_createTaskReference($fileId, $brokerTaskId, $taskName);
        } catch (Exception $e) {
            Logging::error("Invalid request: " . $e->getMessage());
            throw $e;
        }
    }

    /**
     * Create a CeleryTasks object for a pending task
     * TODO: should we have a database layer class to handle Propel operations?
     *
     * @param $fileId       int    CcFiles identifier
     * @param $brokerTaskId int    broker task identifier to so we can asynchronously
     *                             receive completed task messages
     * @param $taskName     string broker task name
     *
     * @return CeleryTasks the created task
     *
     * @throws Exception
     * @throws PropelException
     */
    protected function _createTaskReference($fileId, $brokerTaskId, $taskName) {
        $trackReferenceId = $this->createTrackReference($fileId);
        $task = new CeleryTasks();
        $task->setDbTaskId($brokerTaskId);
        $task->setDbName($taskName);
        $utc = new DateTimeZone("UTC");
        $task->setDbDispatchTime(new DateTime("now", $utc));
        $task->setDbStatus(CELERY_PENDING_STATUS);
        $task->setDbTrackReference($trackReferenceId);
        $task->save();
        return $task;
    }

    /**
     * Update a CeleryTasks object for a completed task
     * TODO: should we have a database layer class to handle Propel operations?
     *
     * @param $task CeleryTasks the completed CeleryTasks object
     * @param $status string    Celery task status
     *
     * @throws Exception
     * @throws PropelException
     */
    public function updateTask($task, $status) {
        $task->setDbStatus($status);
        $task->save();
    }

    /**
     * Update a ThirdPartyTrackReferences object for a completed upload
     *
     * Manipulation and use of the track object is left up to child implementations
     *
     * @param $task     CeleryTasks the completed CeleryTasks object
     * @param $trackId  int         ThirdPartyTrackReferences identifier
     * @param $result   mixed       Celery task result message
     * @param $status   string      Celery task status
     *
     * @return ThirdPartyTrackReferences the updated ThirdPartyTrackReferences object
     *
     * @throws Exception
     * @throws PropelException
     */
    public function updateTrackReference($task, $trackId, $result, $status) {
        static::updateTask($task, $status);
        $ref = ThirdPartyTrackReferencesQuery::create()
            ->findOneByDbId($trackId);
        if (is_null($ref)) {
            $ref = new ThirdPartyTrackReferences();
        }
        $ref->setDbService(static::$_SERVICE_NAME);
        $utc = new DateTimeZone("UTC");
        $ref->setDbUploadTime(new DateTime("now", $utc));
        $ref->save();
        return $ref;
    }

}