Congegni/sintonia
7
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

SAAS-853 - Celery backend for SoundCloud uploads

Browse source
This commit is contained in:
Duncan Sommerville 2015-06-10 15:04:49 -04:00
parent f031d13867
commit 626489bb3b
27 changed files with 813 additions and 250 deletions
Download patch file Download diff file Expand all files Collapse all files

164
airtime_mvc/application/services/ThirdPartyService.php
View file

@ -5,33 +5,79 @@
*/
abstract class ThirdPartyService {
/**
* @var string service access token for accessing remote API
*/
protected $_accessToken;
/**
* @var string service name to store in ThirdPartyTrackReferences database
*/
protected $_SERVICE_NAME = '';
protected $_SERVICE_NAME;
/**
* @var string base URI for third-party tracks
*/
protected $_THIRD_PARTY_TRACK_URI = '';
protected $_THIRD_PARTY_TRACK_URI;
/**
* @var string broker exchange name for third party tasks
*/
protected $_CELERY_EXCHANGE_NAME = 'default';
/**
* @var string celery task name for third party uploads
*/
protected $_CELERY_UPLOAD_TASK_NAME = 'upload';
/**
* @var string status string for pending tasks
*/
protected $_PENDING_STATUS = 'PENDING';
/**
* @var string status string for failed tasks
*/
protected $_FAILED_STATUS = 'FAILED';
/**
* Upload the file with the given identifier to a third-party service
*
* @param int $fileId the local CcFiles identifier
*
* @throws Exception thrown when the upload fails for any reason
*/
abstract function upload($fileId);
public function upload($fileId) {
$file = Application_Model_StoredFile::RecallById($fileId);
$data = array(
'data' => $this->_getUploadData($file),
'token' => $this->_accessToken,
'file_path' => $file->getFilePaths()[0]
);
try {
$brokerTaskId = Application_Model_RabbitMq::sendCeleryMessage($this->_CELERY_UPLOAD_TASK_NAME,
$this->_CELERY_EXCHANGE_NAME,
$data);
$this->_createTaskReference($fileId, $brokerTaskId, $this->_CELERY_UPLOAD_TASK_NAME);
} catch(Exception $e) {
Logging::info("Invalid request: " . $e->getMessage());
// We should only get here if we have an access token, so attempt to refresh
$this->accessTokenRefresh();
}
}
/**
* Create a ThirdPartyTrackReferences and save it to the database
* Create a ThirdPartyTrackReferences object for a pending task
*
* @param $fileId int local CcFiles identifier
* @param $track object third-party service track object
* @param $fileId int local CcFiles identifier
* @param $brokerTaskId int broker task identifier to so we can asynchronously
* receive completed task messages
* @param $taskName string broker task name
*
* @throws Exception
* @throws PropelException
*/
protected function _createTrackReference($fileId, $track) {
protected function _createTaskReference($fileId, $brokerTaskId, $taskName) {
// First, check if the track already has an entry in the database
$ref = ThirdPartyTrackReferencesQuery::create()
->filterByDbService($this->_SERVICE_NAME)
@ -40,9 +86,12 @@ abstract class ThirdPartyService {
$ref = new ThirdPartyTrackReferences();
}
$ref->setDbService($this->_SERVICE_NAME);
$ref->setDbForeignId($track->id);
$ref->setDbBrokerTaskId($brokerTaskId);
$ref->setDbBrokerTaskName($taskName);
$utc = new DateTimeZone("UTC");
$ref->setDbBrokerTaskDispatchTime(new DateTime("now", $utc));
$ref->setDbFileId($fileId);
$ref->setDbStatus($track->state);
$ref->setDbStatus($this->_PENDING_STATUS);
$ref->save();
}
@ -91,6 +140,103 @@ abstract class ThirdPartyService {
return $serviceId > 0 ? $this->_THIRD_PARTY_TRACK_URI . $serviceId : '';
}
/**
* Poll the message queue for this service to see if any tasks with the given name have completed
* If we find any completed tasks, adjust the ThirdPartyTrackReferences table accordingly
* If no task name is passed, we poll all tasks for this service
*
* @param string $taskName the name of the task to poll for
*/
public function pollBrokerTaskQueue($taskName="") {
$pendingTasks = $this->_getPendingTasks($taskName);
foreach ($pendingTasks as $task) {
try {
$result = $this->_getTaskResult($task);
Logging::info(json_decode($result));
$this->_addOrUpdateTrackReference($task->getDbFileId(), json_decode($result));
} catch(CeleryException $e) {
Logging::info("Couldn't retrieve task message for task " . $task->getDbBrokerTaskName()
. " with ID " . $task->getDbBrokerTaskId() . ": " . $e->getMessage());
if ($this->_checkMessageTimeout($task)) {
$task->setDbStatus($this->_FAILED_STATUS);
$task->save();
}
}
}
}
/**
* Return a collection of all pending ThirdPartyTrackReferences to tasks for this service or task
*
* @param string $taskName the name of the task to look for
*
* @return PropelCollection any pending ThirdPartyTrackReferences results for this service
* or task if taskName is provided
*/
protected function _getPendingTasks($taskName) {
$query = ThirdPartyTrackReferencesQuery::create()
->filterByDbService($this->_SERVICE_NAME)
->filterByDbStatus($this->_PENDING_STATUS)
->filterByDbBrokerTaskId('', Criteria::NOT_EQUAL);
if (!empty($taskName)) {
$query->filterByDbBrokerTaskName($taskName);
}
return $query->find();
}
/**
* Get the result from a celery task message in the results queue
* If the task message no longer exists, remove it from the track references table
*
* @param $task ThirdPartyTrackReferences the track reference object
*
* @return array the results from the task message
*
* @throws CeleryException when the result message for this task no longer exists
*/
protected function _getTaskResult($task) {
$message = Application_Model_RabbitMq::getAsyncResultMessage($task->getDbBrokerTaskName(),
$task->getDbBrokerTaskId());
return json_decode($message['body'])->result; // The actual result message from the service
}
/**
* Check if a task message has been unreachable for more our timeout time
*
* @param $task ThirdPartyTrackReferences the track reference object
*
* @return bool true if the dispatch time is empty or it's been more than our timeout time
* since the message was dispatched, otherwise false
*/
protected function _checkMessageTimeout($task) {
$utc = new DateTimeZone("UTC");
$dispatchTime = new DateTime($task->getDbBrokerTaskDispatchTime(), $utc);
$now = new DateTime("now", $utc);
$timeoutSeconds = Application_Model_RabbitMq::$_CELERY_MESSAGE_TIMEOUT / 1000; // Convert from milliseconds
$timeoutInterval = new DateInterval("PT" . $timeoutSeconds . "S");
return (empty($dispatchTime) || $dispatchTime->add($timeoutInterval) <= $now);
}
/**
* Build a parameter array for the file being uploaded to a third party service
*
* @param $file Application_Model_StoredFile the file being uploaded
*
* @return array the track array to send to the third party service
*/
abstract protected function _getUploadData($file);
/**
* Update a ThirdPartyTrackReferences object for a completed task
*
* @param $fileId int local CcFiles identifier
* @param $track object third-party service track object
*
* @throws Exception
* @throws PropelException
*/
abstract protected function _addOrUpdateTrackReference($fileId, $track);
/**
* Check whether an OAuth access token exists for the third-party client
*