diff --git a/livesupport/modules/storageServer/var/BasicStor.php b/livesupport/modules/storageServer/var/BasicStor.php index 3fa06ac60..275de9844 100644 --- a/livesupport/modules/storageServer/var/BasicStor.php +++ b/livesupport/modules/storageServer/var/BasicStor.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.20 $ + Version : $Revision: 1.21 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/BasicStor.php,v $ ------------------------------------------------------------------------------*/ @@ -37,12 +37,14 @@ define('GBERR_AOBJNEX', 46); define('GBERR_NOTF', 47); define('GBERR_SESS', 48); define('GBERR_PREF', 49); +define('GBERR_TOKEN', 50); +define('GBERR_PUT', 51); -define('GBERR_NOTIMPL', 50); +define('GBERR_NOTIMPL', 69); require_once "../../../alib/var/alib.php"; require_once "StoredFile.php"; -#require_once "Transport.php"; +require_once "Transport.php"; /** * BasicStor class @@ -50,7 +52,7 @@ require_once "StoredFile.php"; * Core of LiveSupport file storage module * * @author $Author: tomas $ - * @version $Revision: 1.20 $ + * @version $Revision: 1.21 $ * @see Alib */ class BasicStor extends Alib{ @@ -111,16 +113,17 @@ class BasicStor extends Alib{ * @param mdataFileLP string, local path of metadata file * @param gunid string, global unique id OPTIONAL * @param ftype string, internal file type + * @param mdataLoc string 'file'|'string' (optional) * @return int * @exception PEAR::error */ - function bsPutFile($parid, $fileName, - $mediaFileLP, $mdataFileLP, $gunid=NULL, $ftype='unKnown') + function bsPutFile($parid, $fileName, $mediaFileLP, $mdataFileLP, + $gunid=NULL, $ftype='unKnown', $mdataLoc='file') { $name = "$fileName"; $id = $this->addObj($name , 'File', $parid); $ac =& StoredFile::insert( - &$this, $id, $name, $mediaFileLP, $mdataFileLP, 'file', + &$this, $id, $name, $mediaFileLP, $mdataFileLP, $mdataLoc, $gunid, $ftype ); if(PEAR::isError($ac)){ @@ -236,11 +239,12 @@ class BasicStor extends Alib{ * Delete file * * @param id int, virt.file's local id + * @param forced boolean, unconditional delete * @return true or PEAR::error */ - function bsDeleteFile($id) + function bsDeleteFile($id, $forced=FALSE) { - $res = $this->removeObj($id); + $res = $this->removeObj($id, $forced); return $res; } @@ -353,7 +357,8 @@ class BasicStor extends Alib{ * * @param id int, virt.file's local id * @param part string, 'media'|'metadata' - * @return array with: downloadable URL, download token + * @return array with strings: + * downloadable URL, download token, chsum, size, filename */ function bsOpenDownload($id, $part='media') { @@ -361,18 +366,31 @@ class BasicStor extends Alib{ if(PEAR::isError($ac)) return $ac; $gunid = $ac->gunid; switch($part){ - case"media": - $fname = $ac->_getRealRADFname(); - $ext = $ac->_getExt(); - break; - case"metadata": - $fname = $ac->_getRealMDFname(); - $ext = "xml"; - break; + case"media": + $realfile = $ac->_getRealRADFname(); + $ext = $ac->_getExt(); + $filename = $ac->_getFileName(); + break; + case"metadata": + $realfile = $ac->_getRealMDFname(); + $ext = "xml"; + $filename = $ac->_getFileName(); + break; + default: + return PEAR::raiseError( + "BasicStor::bsOpenDownload: unknown part ($part)" + ); } - $acc = $this->bsAccess($fname, $ext, $gunid, 'download'); + $acc = $this->bsAccess($realfile, $ext, $gunid, 'download'); + if(PEAR::isError($acc)){ return $acc; } $url = $this->getUrlPart()."access/".basename($acc['fname']); - return array('url'=>$url, 'token'=>$acc['token']); + $chsum = md5_file($realfile); + $size = filesize($realfile); + return array( + 'url'=>$url, 'token'=>$acc['token'], + 'chsum'=>$chsum, 'size'=>$size, + 'filename'=>$filename + ); } /** @@ -403,6 +421,10 @@ class BasicStor extends Alib{ { $ext = ''; $token = StoredFile::_createGunid(); + $res = $this->dbc->query(" + DELETE FROM {$this->accessTable} WHERE token=x'$token'::bigint + "); + if(PEAR::isError($res)){ return $res; } $res = $this->dbc->query(" INSERT INTO {$this->accessTable} (gunid, token, ext, chsum, type, ts) @@ -426,9 +448,11 @@ class BasicStor extends Alib{ */ function bsClosePut($token) { + $token = StoredFile::_normalizeGunid($token); if(!$this->bsCheckToken($token, 'put')){ return PEAR::raiseError( - "BasicStor::bsClosePut: invalid token ($token)" + "BasicStor::bsClosePut: invalid token ($token)", + GBERR_TOKEN ); } $chsum = $this->dbc->getOne(" @@ -444,12 +468,47 @@ class BasicStor extends Alib{ if($chsum != $md5sum){ if(file_exists($fname)) @unlink($fname); return PEAR::raiseError( - "BasicStor::bsClosePut: md5sum does not match (token=$token)" + "BasicStor::bsClosePut: md5sum does not match (token=$token)", + GBERR_PUT ); } return $fname; } + /** + * Check uploaded file + * + * @param token string, put token + * @return hash, ( + * status: boolean, + * size: int - filesize + * expectedsum: string - expected checksum + * realsum: string - checksum of uploaded file + * ) + */ + function bsCheckPut($token) + { + if(!$this->bsCheckToken($token, 'put')){ + return PEAR::raiseError( + "BasicStor::bsClosePut: invalid token ($token)" + ); + } + $chsum = $this->dbc->getOne(" + SELECT chsum FROM {$this->accessTable} + WHERE token=x'{$token}'::bigint + "); + if(PEAR::isError($chsum)){ return $chsum; } + $fname = "{$this->accessDir}/$token"; + $md5sum = md5_file($fname); + $size = filesize($fname); + $status = ($chsum == $md5sum); + return array( + 'status'=>$status, 'size'=>$size, + 'expectedsum'=>$chsum, + 'realsum'=>$md5sum, + ); + } + /** * Return starting part of storageServer URL * @@ -644,7 +703,7 @@ class BasicStor extends Alib{ if($v['type'] == 'File'){ $gunid = $this->_gunidFromId($v['id']); $listArr[$i]['type'] = - StoredFile::_getType($gunid); + $this->_getType($gunid); $listArr[$i]['gunid'] = $gunid; if(StoredFIle::_getState($gunid) == 'incomplete') unset($listArr[$i]); @@ -725,12 +784,47 @@ class BasicStor extends Alib{ $type = $this->getObjName($oid, 'type'); if($type == 'File'){ $ftype = - StoredFile::_getType($this->_gunidFromId($oid)); + $this->_getType($this->_gunidFromId($oid)); if(!is_null($ftype)) $type=$ftype; } return $type; } + /** + * Add new user with home folder + * + * @param login string + * @param pass string OPT + * @return int/err + */ + function addSubj($login, $pass=NULL) + { + $uid = parent::addSubj($login, $pass); + if(PEAR::isError($uid)) return $uid; + $fid = $this->bsCreateFolder($this->storId, $login); + if(PEAR::isError($fid)) return $fid; + $res = $this->addPerm($uid, '_all', $fid, 'A'); + if(PEAR::isError($res)) return $res; + return $uid; + } + /** + * Remove user and his home folder + * + * @param login string + * @param uid int OPT + * @return boolean/err + */ + function removeSubj($login, $uid=NULL) + { + $res = parent::removeSubj($login, $pass); + if(PEAR::isError($res)) return $res; + $id = $this->getObjId($login, $this->storId); + if(PEAR::isError($id)) return $id; + $res = $this->bsDeleteFile($id); + if(PEAR::isError($res)) return $res; + return TRUE; + } + /* ==================================================== "private" methods */ /** * Check authorization - auxiliary method @@ -767,6 +861,10 @@ class BasicStor extends Alib{ $parid = $this->getObjId( $this->getSessLogin($sessid), $this->storId ); + if(is_null($parid)){ + return PEAR::raiseError("BasicStor::_getHomeDirId: ". + "homedir not found", GBERR_NOTF); + } return $parid; } @@ -801,6 +899,21 @@ class BasicStor extends Alib{ return StoredFile::_normalizeGunid($gunid); } + /** + * Get storage-internal file type + * + * @param gunid string, global unique id of file + * @return string, see install() + */ + function _getType($gunid) + { + $ftype = $this->dbc->getOne(" + SELECT ftype FROM {$this->filesTable} + WHERE gunid=x'$gunid'::bigint + "); + return $ftype; + } + /* ------------------------------------------ redefined "private" methods */ /** * Copy virtual file.
@@ -845,21 +958,22 @@ class BasicStor extends Alib{ * Redefined from parent class. * * @param id int, local id of removed object + * @param forced boolean, unconditional delete * @return true or PEAR::error */ - function removeObj($id) + function removeObj($id, $forced=FALSE) { switch($ot = $this->getObjType($id)){ case"audioclip": case"playlist": $ac =& StoredFile::recall(&$this, $id); if(PEAR::isError($ac)) return $ac; - if($ac->isEdited()){ + if($ac->isEdited() && !$forced){ return PEAR::raiseError( 'BasicStor.php: removeObj: is edited' ); } - if($ac->isAccessed()){ + if($ac->isAccessed() && !$forced){ return PEAR::raiseError( 'BasicStor.php: removeObj: is accessed' ); @@ -923,7 +1037,7 @@ class BasicStor extends Alib{ // $this->dbc->query("DELETE FROM {$this->filesTable}"); $ids = $this->dbc->getAll("SELECT id FROM {$this->filesTable}"); if(is_array($ids)) foreach($ids as $i=>$item){ - $this->bsDeleteFile($item['id']); + $this->bsDeleteFile($item['id'], TRUE); } parent::deleteData(); $this->initData(); diff --git a/livesupport/modules/storageServer/var/GreenBox.php b/livesupport/modules/storageServer/var/GreenBox.php index 93391bd19..6ebcb1ae6 100644 --- a/livesupport/modules/storageServer/var/GreenBox.php +++ b/livesupport/modules/storageServer/var/GreenBox.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.23 $ + Version : $Revision: 1.24 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/GreenBox.php,v $ ------------------------------------------------------------------------------*/ @@ -35,7 +35,7 @@ require_once "BasicStor.php"; * LiveSupport file storage module * * @author $Author: tomas $ - * @version $Revision: 1.23 $ + * @version $Revision: 1.24 $ * @see BasicStor */ class GreenBox extends BasicStor{ @@ -278,41 +278,6 @@ class GreenBox extends BasicStor{ /* ---------------------------------------------------- redefined methods */ - /** - * Add new user with home folder - * - * @param login string - * @param pass string OPT - * @return int/err - */ - function addSubj($login, $pass=NULL) - { - $uid = parent::addSubj($login, $pass); - if(PEAR::isError($uid)) return $uid; - $fid = $this->bsCreateFolder($this->storId, $login); - if(PEAR::isError($fid)) return $fid; - $res = $this->addPerm($uid, '_all', $fid, 'A'); - if(PEAR::isError($res)) return $res; - return $uid; - } - /** - * Remove user and his home folder - * - * @param login string - * @param uid int OPT - * @return boolean/err - */ - function removeSubj($login, $uid=NULL) - { - $res = parent::removeSubj($login, $pass); - if(PEAR::isError($res)) return $res; - $id = $this->getObjId($login, $this->storId); - if(PEAR::isError($id)) return $id; - $res = $this->bsDeleteFile($id); - if(PEAR::isError($res)) return $res; - return TRUE; - } - /** * Get file's path in virtual filesystem * diff --git a/livesupport/modules/storageServer/var/LocStor.php b/livesupport/modules/storageServer/var/LocStor.php index 2641f363b..754596618 100644 --- a/livesupport/modules/storageServer/var/LocStor.php +++ b/livesupport/modules/storageServer/var/LocStor.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.22 $ + Version : $Revision: 1.23 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/LocStor.php,v $ ------------------------------------------------------------------------------*/ @@ -48,7 +48,9 @@ class LocStor extends BasicStor{ * @param chsum string, md5 checksum of media file * @return struct {url:writable URL for HTTP PUT, token:access token */ - function storeAudioClipOpen($sessid, $gunid, $metadata, $fname, $chsum) + function storeAudioClipOpen( + $sessid, $gunid, $metadata, $fname, $chsum, $ftype='audioclip' + ) { // test if specified gunid exists: if(!preg_match("|^([0-9a-fA-F]{16})?$|", $gunid)){ @@ -59,8 +61,9 @@ class LocStor extends BasicStor{ $ac =& StoredFile::recallByGunid(&$this, $gunid); if(!PEAR::isError($ac)){ // gunid exists - do replace + $oid = $ac->getId(); if(($res = $this->_authorize( - 'write', $ac->getId(), $sessid + 'write', $oid, $sessid )) !== TRUE) return $res; if($ac->isAccessed()){ return PEAR::raiseError( @@ -68,7 +71,7 @@ class LocStor extends BasicStor{ ); } $res = $ac->replace( - $ac->getId(), $ac->name, '', $metadata, 'string' + $oid, $ac->name, '', $metadata, 'string' ); if(PEAR::isError($res)) return $res; }else{ @@ -82,7 +85,7 @@ class LocStor extends BasicStor{ if(PEAR::isError($oid)) return $oid; $ac =& StoredFile::insert( &$this, $oid, '', '', $metadata, 'string', - $gunid, 'audioclip' + $gunid, $ftype ); if(PEAR::isError($ac)){ $res = $this->removeObj($oid); @@ -121,6 +124,17 @@ class LocStor extends BasicStor{ return $ac->gunid; } + /** + * Check uploaded file + * + * @param token string, put token + * @return hash, (status: boolean, size: int - filesize) + */ + function uploadCheck($token) + { + return $this->bsCheckPut($token); + } + /* --------------------------------------------------------------- access */ /** * Make access to audio clip @@ -158,16 +172,18 @@ class LocStor extends BasicStor{ * * @param sessid string, session id * @param gunid string, global unique id - * @return array with: downloadable URL, download token + * @return array with strings: + * downloadable URL, download token, chsum, size, filename */ function downloadRawAudioDataOpen($sessid, $gunid) { - $res = $this->existsAudioClip($sessid, $gunid); - if(PEAR::isError($res)) return $res; + $ex = $this->existsAudioClip($sessid, $gunid); + if(PEAR::isError($ex)) return $ex; $id = $this->_idFromGunid($gunid); - if(is_null($id)){ + if(is_null($id) || !$ex){ return PEAR::raiseError( - "LocStor::downloadRawAudioDataOpen: gunid not found ($gunid)" + "LocStor::downloadRawAudioDataOpen: gunid not found ($gunid)", + GBERR_NOTF ); } if(($res = $this->_authorize('read', $id, $sessid)) !== TRUE) @@ -191,12 +207,13 @@ class LocStor extends BasicStor{ * * @param sessid string, session id * @param gunid string, global unique id - * @return array with: downloadable URL, download token + * @return array with strings: + * downloadable URL, download token, chsum, filename */ function downloadMetadataOpen($sessid, $gunid) { - $res = $this->existsAudioClip($sessid, $gunid); - if(PEAR::isError($res)) return $res; +# $res = $this->existsAudioClip($sessid, $gunid); +# if(PEAR::isError($res)) return $res; $id = $this->_idFromGunid($gunid); if(is_null($id)){ return PEAR::raiseError( @@ -205,7 +222,9 @@ class LocStor extends BasicStor{ } if(($res = $this->_authorize('read', $id, $sessid)) !== TRUE) return $res; - return $this->bsOpenDownload($id, 'metadata'); + $res = $this->bsOpenDownload($id, 'metadata'); + #unset($res['filename']); + return $res; } /** @@ -350,7 +369,7 @@ class LocStor extends BasicStor{ default: return $ac; } } - if(!is_null($ftype) && ($ac->_getType() != $ftype)) return FALSE; + if(!is_null($ftype) && ($this->_getType($gunid) != $ftype)) return FALSE; if(($res = $this->_authorize('read', $ac->getId(), $sessid)) !== TRUE) return $res; return TRUE; @@ -399,6 +418,10 @@ class LocStor extends BasicStor{ function resetStorage($input='') { $this->deleteData(); + if(!$this->config['isArchive']){ + $tr =& new Transport($this->dbc, $this, $this->config); + $tr->resetData(); + } $rootHD = $this->getObjId('root', $this->storId); include"../tests/sampleData.php"; $res = array('audioclips'=>array(), 'playlists'=>array()); @@ -483,7 +506,8 @@ class LocStor extends BasicStor{ * * @param sessid string, session ID * @param playlistId string, playlist global unique ID - * @return struct {url:readable URL for HTTP GET, token:access token} + * @return struct + * {url:readable URL for HTTP GET, token:access token, chsum:checksum} */ function editPlaylist($sessid, $playlistId) { @@ -505,6 +529,7 @@ class LocStor extends BasicStor{ $res = $this->bsOpenDownload($id, 'metadata'); if(PEAR::isError($res)){ return $res; } $this->_setEditFlag($playlistId, TRUE); + unset($res['filename']); return $res; } @@ -557,7 +582,8 @@ class LocStor extends BasicStor{ * * @param sessid string, session ID * @param playlistId string, playlist global unique ID - * @return struct {url:readable URL for HTTP GET, token:access token + * @return struct + * {url:readable URL for HTTP GET, token:access token, chsum:checksum} */ function accessPlaylist($sessid, $playlistId) { @@ -565,13 +591,16 @@ class LocStor extends BasicStor{ if(PEAR::isError($ex)){ return $ex; } if(!$ex){ return PEAR::raiseError( - 'LocStor.php: accessPlaylist: playlist not found' + "LocStor.php: accessPlaylist: playlist not found ($playlistId)", + GBERR_NOTF ); } $id = $this->_idFromGunid($playlistId); if(($res = $this->_authorize('read', $id, $sessid)) !== TRUE) return $res; - return $this->bsOpenDownload($id, 'metadata'); + $res = $this->bsOpenDownload($id, 'metadata'); + unset($res['filename']); + return $res; } /** diff --git a/livesupport/modules/storageServer/var/StoredFile.php b/livesupport/modules/storageServer/var/StoredFile.php index 9884116ff..1b49b1e1c 100644 --- a/livesupport/modules/storageServer/var/StoredFile.php +++ b/livesupport/modules/storageServer/var/StoredFile.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.16 $ + Version : $Revision: 1.17 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/StoredFile.php,v $ ------------------------------------------------------------------------------*/ @@ -93,7 +93,7 @@ class StoredFile{ $ac->mime = "unKnown"; $emptyState = TRUE; if($ac->name=='') $ac->name=$ac->gunid; - $this->dbc->query("BEGIN"); + $ac->dbc->query("BEGIN"); $res = $ac->dbc->query(" INSERT INTO {$ac->filesTable} (id, name, gunid, mime, state, ftype) @@ -101,7 +101,7 @@ class StoredFile{ ('$oid', '{$ac->name}', x'{$ac->gunid}'::bigint, '{$ac->mime}', 'incomplete', '$ftype') "); - if(PEAR::isError($res)){ $this->dbc->query("ROLLBACK"); return $res; } + if(PEAR::isError($res)){ $ac->dbc->query("ROLLBACK"); return $res; } // --- metadata insert: if($metadata != ''){ if($mdataLoc=='file' && !file_exists($metadata)) @@ -111,7 +111,7 @@ class StoredFile{ } $res = $ac->md->insert($metadata, $mdataLoc); if(PEAR::isError($res)){ - $this->dbc->query("ROLLBACK"); return $res; + $ac->dbc->query("ROLLBACK"); return $res; } $emptyState = FALSE; } @@ -124,7 +124,7 @@ class StoredFile{ } $res = $ac->rmd->insert($mediaFileLP); if(PEAR::isError($res)){ - $this->dbc->query("ROLLBACK"); return $res; + $ac->dbc->query("ROLLBACK"); return $res; } $mime = $ac->rmd->getMime(); //$gb->debugLog("gunid={$ac->gunid}, mime=$mime"); @@ -140,8 +140,8 @@ class StoredFile{ $res = $ac->setState('ready'); if(PEAR::isError($res)){ $ac->dbc->query("ROLLBACK"); return $res; } } - $res = $this->dbc->query("COMMIT"); - if(PEAR::isError($res)){ $this->dbc->query("ROLLBACK"); return $res; } + $res = $ac->dbc->query("COMMIT"); + if(PEAR::isError($res)){ $ac->dbc->query("ROLLBACK"); return $res; } return $ac; } @@ -224,7 +224,7 @@ class StoredFile{ { $ac =& StoredFile::insert( &$src->gb, $nid, $src->name, $src->_getRealRADFname(), - '', '', NULL, $src->_getType() + '', '', NULL, $src->gb->_getType($src->gunid) ); if(PEAR::isError($ac)) return $ac; $ac->md->replace($src->md->getMetaData(), 'string'); @@ -256,7 +256,7 @@ class StoredFile{ $this->dbc->query("ROLLBACK"); return $res; } if($metadata != ''){ // metadata - $res = $this->md->replace($metadata, $mdataLoc); + $res = $this->replaceMetaData($metadata, $mdataLoc); }else{ $res = $this->md->delete(); } @@ -495,6 +495,7 @@ class StoredFile{ SELECT to_hex(gunid) FROM {$this->filesTable} WHERE gunid=x'{$this->gunid}'::bigint "); + if(PEAR::isError($indb)) return $indb; return (!is_null($indb) && $this->rmd->exists()); } @@ -585,22 +586,6 @@ class StoredFile{ "); } - /** - * Get storage-internal file type - * - * @param gunid string, optional, global unique id of file - * @return string, see install() - */ - function _getType($gunid=NULL) - { - if(is_null($gunid)) $gunid = $this->gunid; - $ftype = $this->dbc->getOne(" - SELECT ftype FROM {$this->filesTable} - WHERE gunid=x'$gunid'::bigint - "); - return $ftype; - } - /** * Get storage-internal file state * @@ -639,8 +624,9 @@ class StoredFile{ function _getResDir() { $resDir="{$this->gb->storageDir}/".substr($this->gunid, 0, 3); + #$this->gb->debugLog("$resDir"); // see Transport::_getResDir too for resDir name create code - if(!file_exists($resDir)){ mkdir($resDir, 02775); chmod($resDir, 02775); } + if(!is_dir($resDir)){ mkdir($resDir, 02775); chmod($resDir, 02775); } return $resDir; } @@ -671,6 +657,7 @@ class StoredFile{ */ function _getAccessFname($token, $ext='EXT') { + $token = StoredFile::_normalizeGunid($token); return "{$this->accessDir}/$token.$ext"; } } diff --git a/livesupport/modules/storageServer/var/Transport.php b/livesupport/modules/storageServer/var/Transport.php index 4ee7b402a..96c91e3fd 100644 --- a/livesupport/modules/storageServer/var/Transport.php +++ b/livesupport/modules/storageServer/var/Transport.php @@ -23,11 +23,15 @@ Author : $Author: tomas $ - Version : $Revision: 1.3 $ + Version : $Revision: 1.4 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/Transport.php,v $ ------------------------------------------------------------------------------*/ -include_once "xmlrpc/XML/RPC.php"; +define('TRERR_', 70); +define('TRERR_MD', 71); +define('TRERR_TOK', 72); + +include_once "XML/RPC.php"; /** * Class for handling file tranport between StorageServer and ArchiveServer
@@ -43,29 +47,703 @@ class Transport{ * Constructor * * @param dbc PEAR DB object reference + * @param gb LocStor object reference * @param config config array */ - function Transport(&$dbc, $config) + function Transport(&$dbc, &$gb, $config) { $this->dbc =& $dbc; $this->config = $config; $this->transTable = $config['tblNamePrefix'].'trans'; $this->transDir = $config['transDir']; + $this->gb =& $gb; } + /* ================================================== status info methods */ /** * Return state of transport job * + * @param trtok string, transport token + * @return string, transport state */ - function getTransportStatus($trid) + function getState($trtok) { $row = $this->dbc->getRow( - "SELECT state FROM {$this->transTable} WHERE trid='$trid'" + "SELECT state FROM {$this->transTable} WHERE trtok='$trtok'" ); - if(PEAR::isError($res)) return $res; + if(PEAR::isError($row)) return $row; + if(is_null($row)){ + return PEAR::raiseError("Transport::getState:". + " invalid transport token ($trtok)", TRERR_TOK + ); + } return $row['state']; } + /** + * Return hash with useful information about transport + * + * @param trtok string, transport token + * @return hash: + * trtype, + * direction, + * status, + * expectedsize, + * realsize, + * expectedsum, + * realsum + */ + function getTransportInfo($trtok) + { + $row = $this->dbc->getRow(" + SELECT + trtype, direction, state, + expectedsize, realsize, + expectedsum, realsum + FROM {$this->transTable} + WHERE trtok='$trtok' + "); + if(PEAR::isError($row)){ return $row; } + if(is_null($row)){ + return PEAR::raiseError("Transport::getTransportInfo:". + " invalid transport token ($trtok)", TRERR_TOK + ); + } + $row['status'] == ($row['state'] == 'closed'); +# unset($row['state']); + return $row; + } + + /* ======================================================= upload methods */ + + /** + * + */ + function uploadToArchive($gunid, $sessid='') + { + $trtype = $this->gb->_getType($gunid); + switch($trtype){ + case"audioclip": + $ex = $this->gb->existsAudioClip($sessid, $gunid); + break; + case"playlist": + $ex = $this->gb->existsPlaylist($sessid, $gunid); + break; + default: + return PEAR::raiseError("Transport::uploadToArchive:". + " unknown trtype ($trtype)" + ); + } + if(PEAR::isError($ex)) return $ex; + if(!$ex){ + return PEAR::raiseError("Transport::uploadToArchive:". + " $trtype not found ($gunid)" + ); + } + $trtok = $this->_createTrtok(); + $id = $this->dbc->nextId("{$this->transTable}_id_seq"); + $res = $this->dbc->query(" + INSERT INTO {$this->transTable} + (id, trtok, direction, state, trtype, + gunid, + expectedsum, realsum, expectedsize, realsize + ) + VALUES + ($id, '$trtok', 'up', 'init', '$trtype', + x'$gunid'::bigint, + '?e', '?r', '0', '0' + ) + "); + if(PEAR::isError($res)) return $res; +#?? $this->uploadCron(); + return $trtok; + } + + /** + * + */ + function uploadCron() + { + // fetch all opened uploads + $rows = $this->dbc->getAll(" + SELECT + id, trtok, state, trtype, + to_hex(gunid)as gunid, to_hex(pdtoken)as pdtoken, + fname, localfile, expectedsum, expectedsize, url + FROM {$this->transTable} + WHERE direction='up' AND state<>'closed' + "); + if(PEAR::isError($rows)){ $this->trLogPear($rows); return FALSE; } + if(count($rows)==0) return TRUE; + // login to archive server + $r = $this->loginToArchive(); + if(PEAR::isError($r)){ $this->trLog("Login error"); return FALSE; } + $asessid = $r['sessid']; + chdir($this->config['transDir']); + // for all opened uploads: + foreach($rows as $i=>$row){ + $row['pdtoken'] = StoredFile::_normalizeGunid($row['pdtoken']); + $row['gunid'] = StoredFile::_normalizeGunid($row['gunid']); + #var_dump($row); + switch($row['state']){ + case"init": // ------ new uploads + $ret = $this->uploadCronInit($row, $asessid); + if(PEAR::isError($ret)){ + $this->trLogPear($ret, $row); break; + } + $row = array_merge($row, $ret); + // break; + case"pending": // ------ pending uploads + $ret = $this->uploadCronPending($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + case"finished": // ------ finished uploads + $ret = $this->uploadCronFinished($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + case"failed": // ------ failed uploads + $ret = $this->uploadCronFailed($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + default: + $this->trLog("uploadCron: unknown state ". + "({$row['state']}, id={$row['id']})"); + } // switch state + } // foreach opened + $r = $this->logoutFromArchive($asessid); + if(PEAR::isError($r)) $this->trLog("Logout error"); + return TRUE; + } + + /** + * + */ + function uploadCronInit($row, $asessid) + { + $this->trLog("INIT UP id={$row['id']}, trtok={$row['trtok']}"); + switch($row['trtype']){ + case"audioclip": + $ac =& StoredFile::recallByGunid(&$this->gb, $row['gunid']); + if(PEAR::isError($ac)) return $ac; + $fpath = $ac->_getRealRADFname(); + $fname = $ac->_getFileName(); + $chsum = $this->_chsum($fpath); + $size = filesize($fpath); + $metadata = file_get_contents($ac->_getRealMDFname()); + $ret = $this->xmlrpcCall( 'archive.storeAudioClipOpen', + array( + 'sessid' => $asessid, + 'gunid' => $row['gunid'], + 'metadata' => $metadata, + 'fname' => $fname, + 'chsum' => $chsum, + ) + ); + if(PEAR::isError($ret)) return $ret; + $r = $this->dbc->query(" + UPDATE {$this->transTable} + SET + state = 'pending', + fname = '$fname', + localfile = '$fpath', + expectedsum = '$chsum', + expectedsize= $size, + url = '{$ret['url']}', + pdtoken = x'{$ret['token']}'::bigint + WHERE id={$row['id']} + "); + if(PEAR::isError($r)) return $r; + return array( + 'state'=>'pending', + 'url'=>$ret['url'], 'pdtoken'=>$ret['token'], + 'localfile'=>$fpath, 'fname'=>$fname, + 'expectedsum'=>$chsum, 'expectedsize'=>$size, + ); + break; + case"playlist": + $ac =& StoredFile::recallByGunid(&$this->gb, $row['gunid']); + if(PEAR::isError($ac)) return $ac; + $fname = $ac->_getFileName(); + $size = filesize($fpath); + $metadata = file_get_contents($ac->_getRealMDFname()); + $ret = $this->xmlrpcCall( 'archive.createPlaylist', + array( + 'sessid' => $asessid, + 'plid' => $row['gunid'], + 'fname' => $fname, + ) + ); + if(PEAR::isError($ret)) return $ret; + $ret = $this->xmlrpcCall( 'archive.editPlaylist', + array( + 'sessid' => $asessid, + 'plid' => $row['gunid'], + 'metadata' => $metadata, + ) + ); + if(PEAR::isError($ret)) return $ret; +# $this->trLog("INIT UP after edit {$ret['token']}"); + $r = $this->xmlrpcCall( 'archive.savePlaylist', + array( + 'sessid' => $asessid, + 'token' => $ret['token'], + 'newPlaylist' => $metadata, + ) + ); + if(PEAR::isError($r)) return $r; +# $this->trLog("INIT UP after save {$r['plid']}"); + $r = $this->dbc->query(" + UPDATE {$this->transTable} + SET + state = 'closed', + fname = '$fname', + url = '{$ret['url']}' + WHERE id={$row['id']} + "); + if(PEAR::isError($r)) return $r; + return array( + 'state'=>'closed', + 'url'=>$ret['url'], 'pdtoken'=>$ret['token'], + 'fname'=>$fname, + ); + break; + case"searchjob": + return array(); + default: + return PEAR::raiseError("Transport::uploadCronInit:". + " unknown trtype ({$row['trtype']})" + ); + } // switch $row['trtype'] + } + + /** + * + */ + function uploadCronPending($row, $asessid) + { + if($row['trtype'] != 'audioclip') return; + $check = $this->uploadCheck($row['pdtoken']); + if(PEAR::isError($check)) return $check; + #var_dump($row); + #var_dump($check); + // test filesize + if(intval($check['size']) < $row['expectedsize']){ + // not finished - upload next part + $command = + "curl -s -C {$check['size']} --max-time 600". + " --speed-time 20 --speed-limit 500". + " --connect-timeout 20". + " -T {$row['localfile']} {$row['url']}"; +# echo "$command\n"; + $res = system($command, $status); + }else{ + // hmmm - we are finished? OK - continue + $status = 0; + } + // status 18 - Partial file. Only a part of the file was transported. + if($status == 0 || $status == 18){ + $check = $this->uploadCheck($row['pdtoken']); + if(PEAR::isError($check)) return $check; + #var_dump($check); + // test checksum + if($check['status'] == TRUE){ + // finished + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET state ='finished', + realsum ='{$check['realsum']}', + realsize={$check['size']} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + }else{ + if(intval($check['size']) < $row['expectedsize']){ + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET + realsum ='{$check['realsum']}', + realsize={$check['realsize']} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + }else{ + // wrong md5 at finish - TODO: start again + // $this->xmlrpcCall('archive.uploadReset', array()); + return PEAR::raiseError("Transport::uploadCron:". + " file uploaded with bad md5 ". + "({$check['realsum']}/{$check['expectedsum']})" + ); + } + } + } + } + + /** + * + */ + function uploadCronFinished($row, $asessid) + { + if($row['trtype'] != 'audioclip') return; + $res = $this->xmlrpcCall( + 'archive.storeAudioClipClose', + array('sessid'=>$asessid, 'token'=>$row['pdtoken']) + ); +# if(PEAR::isError($res)) return $res; + if(PEAR::isError($res)){ + switch($res->getCode()){ + case GBERR_PUT: + // mark as failed + $this->dbc->query(" + UPDATE {$this->transTable} SET state='failed' + WHERE id='{$row['id']}' + "); + break; + return FALSE; + default: +# echo $res->getCode()."\n"; exit; + return $res; + } + } + // close upload in db + $r = $this->dbc->query(" + UPDATE {$this->transTable} SET state='closed' + WHERE id='{$row['id']}' + "); + /* + $r = $this->dbc->query(" + DELETE FROM {$this->transTable} + WHERE id='{$row['id']}' + "); + */ + if(PEAR::isError($r)) return $r; + $this->trLog("FIN UP id={$row['id']}, trtok={$row['trtok']}". + "\n ".serialize($row)); + } + + /** + * + */ + function uploadCronFailed($row, $asessid) + { + /* + $r = $this->dbc->query(" + DELETE FROM {$this->transTable} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($r)) return $r; + */ + } + + /** + * Check state of uploaded file + * + * @param pdtoken string, put token + * @return hash: chsum, size, url + */ + function uploadCheck($pdtoken) + { + $ret = $this->xmlrpcCall( + 'archive.uploadCheck', + array('token'=>$pdtoken) + ); + return $ret; + } + + /* ===================================================== download methods */ + /** + * + */ + function downloadFromArchive($gunid, $sessid='') + { + // insert transport record to db + $uid = $this->gb->getSessUserId($sessid); + if(PEAR::isError($uid)) return $uid; + if(is_null($uid)){ + return PEAR::raiseError("Transport::downloadFromArchive: ". + "invalid session id",GBERR_SESS); + } + $parid = $this->gb->_getHomeDirId($sessid); + if(PEAR::isError($parid)) return $parid; + $trtok = $this->_createTrtok(); + $id = $this->dbc->nextId("{$this->transTable}_id_seq"); + if(PEAR::isError($id)) return $id; + $res = $this->dbc->query(" + INSERT INTO {$this->transTable} + (id, trtok, direction, state, trtype, + gunid, + expectedsum, realsum, expectedsize, realsize, + uid, parid + ) + VALUES + ($id, '$trtok', 'down', 'init', '?', + x'$gunid'::bigint, + '?e', '?r', '0', '0', + $uid, $parid + ) + "); + if(PEAR::isError($res)) return $res; +#?? $this->downloadCron(); + return $trtok; + } + + /** + * + */ + function downloadCron() + { + // fetch all opened downloads + $rows = $this->dbc->getAll(" + SELECT + id, trtok, state, trtype, + to_hex(gunid)as gunid, to_hex(pdtoken)as pdtoken, + fname, localfile, expectedsum, expectedsize, url, + uid, parid + FROM {$this->transTable} + WHERE direction='down' AND state<>'closed' + "); + if(PEAR::isError($rows)){ $this->trLogPear($rows); return FALSE; } + if(count($rows)==0) return TRUE; + // login to archive server + $r = $this->loginToArchive(); + if(PEAR::isError($r)){ $this->trLog("Login error"); return FALSE; } + $asessid = $r['sessid']; + chdir($this->config['transDir']); + // for all opened downloads: + foreach($rows as $i=>$row){ + $row['pdtoken'] = StoredFile::_normalizeGunid($row['pdtoken']); + $row['gunid'] = StoredFile::_normalizeGunid($row['gunid']); + switch($row['state']){ + case"init": // ------ new downloads + $ret = $this->downloadCronInit($row, $asessid); + if(PEAR::isError($ret)){ + $this->trLogPear($ret, $row); break; + } + $row = array_merge($row, $ret); + #break; + case"pending": // ------ pending downloads + $ret = $this->downloadCronPending($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + case"finished": // ------ finished downloads + $ret = $this->downloadCronFinished($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + case"failed": // ------ failed downloads + $ret = $this->downloadCronFailed($row, $asessid); + if(PEAR::isError($ret)){ $this->trLogPear($ret, $row); } + break; + default: + $this->trLog("downloadCron: unknown state ". + "({$row['state']}, id={$row['id']})"); + } // switch state + } // foreach opened + $r = $this->logoutFromArchive($asessid); + if(PEAR::isError($r)) $this->trLog("Logout error"); + return TRUE; + } + + /** + * + */ + function downloadCronInit($row, $asessid) + { + $ret = $this->xmlrpcCall('archive.downloadRawAudioDataOpen', + array('sessid'=>$asessid, 'gunid'=>$row['gunid']) + ); + if(PEAR::isError($ret)){ + // catch 'not found' exception: + if($ret->getCode() != 847) return $ret; + else $trtype = '?'; + }else $trtype = 'audioclip'; + if($trtype == '?'){ + $ret = $this->xmlrpcCall('archive.existsPlaylist', + array('sessid'=>$asessid, 'plid'=>$row['gunid']) + ); + if(PEAR::isError($ret)){ + // catch 'not found' exception: + if($ret->getCode() != 847) return $ret; + else $trtype = '?'; + }else{ + $trtype = 'playlist'; + $r1 = $this->downloadMetadata($row['gunid'], $asessid); + if(PEAR::isError($r1)) return $r1; + $r2 = $this->gb->bsPutFile($row['parid'], $r1['filename'], + '', $r1['mdata'], $row['gunid'], $trtype, 'string'); + if(PEAR::isError($r2)) return $r2; + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET state='closed', + trtype='$trtype' + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + return array( + 'state'=>'closed', 'trtype'=>$trtype, + 'fname'=>$r1['filename'], + ); + } + } + if($trtype == '?'){ + return PEAR::raiseError("Transport::downloadCronInit:". + " unknown trtype ({$row['trtype']})" + ); + } + $fpath = $this->transDir."/".basename($ret['url']); // *** + touch($fpath); + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET + state = 'pending', + trtype = '$trtype', + fname = '{$ret['filename']}', + localfile = '$fpath', + expectedsum = '{$ret['chsum']}', + expectedsize= '{$ret['size']}', + url = '{$ret['url']}', + pdtoken = x'{$ret['token']}'::bigint + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + $this->trLog("INIT DOWN id={$row['id']}, trtok={$row['trtok']}"); + return array( + 'state'=>'pending', 'trtype'=>$trtype, + 'url'=>$ret['url'], 'pdtoken'=>$ret['token'], + 'localfile'=>$fpath, 'fname'=>$ret['filename'], + 'expectedsum'=>$ret['chsum'], 'expectedsize'=>$ret['size'], + ); + } + + /** + * + */ + function downloadCronPending($row, $asessid) + { + if($row['trtype'] != 'audioclip') return; + // wget the file + $command = + "wget -q -c --timeout={$this->timeout}". + " --waitretry={$this->waitretry}". + " -t {$this->retries} {$row['url']}"; +# echo "$command\n"; + $res = system($command, $status); + // check consistency + $chsum = $this->_chsum($row['localfile']); + $size = filesize($row['localfile']); + if($status == 0){ + if($chsum == $row['expectedsum']){ + // mark download as finished + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET state='finished', + realsum ='{$chsum}', + realsize={$size} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + }else{ + @unlink($row['localfile']); + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET + realsum ='{$chsum}', + realsize={$size} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($res)) return $res; + } + } + } + + /** + * + */ + function downloadCronFinished($row, $asessid) + { + if($row['trtype'] != 'audioclip') return; + // call archive that downloads have been finished OK + $res = $this->xmlrpcCall( + 'archive.downloadRawAudioDataClose', + array('sessid'=>$asessid, 'token'=>$row['pdtoken']) + ); + if(PEAR::isError($res)) return $res; + $res2 = $this->downloadMetadata($row['gunid'], $asessid); + if(PEAR::isError($res2)) return $res2; + $mdata = $res2['mdata']; + $name = $row['fname']; + $this->trLog("FIN1 DOWN id={$row['id']}, trtok={$row['trtok']}". + "\n ".serialize($row)); + $ac =& StoredFile::recallByGunid(&$this->gb, $row['gunid']); + if(!PEAR::isError($ac)){ + // gunid exists - do replace + $id = $ac->getId(); + $ac->replace( + $id, $name, $row['localfile'], $mdata, 'string' + ); + if(PEAR::isError($ac)) return $ac; + }else{ + // gunid doesn't exists - do insert + $id = $this->gb->addObj($name , 'File', $row['parid']); + if(PEAR::isError($id)) return $id; + $ac =& StoredFile::insert( + &$this->gb, $id, $name, $row['localfile'], $mdata, 'string', + $row['gunid'], 'audioclip' + ); + if(PEAR::isError($ac)) return $ac; + } + // close download in db + $res = $this->dbc->query(" + UPDATE {$this->transTable} + SET state='closed' + WHERE id='{$row['id']}' + "); + /* + $res = $this->dbc->query(" + DELETE FROM {$this->transTable} + WHERE id='{$row['id']}' + "); + */ + if(PEAR::isError($res)) return $res; + $this->trLog("FIN DOWN id={$row['id']}, trtok={$row['trtok']}". + "\n ".serialize($row)); + } + + /** + * + */ + function downloadCronFailed($row, $asessid) + { + /* + $r = $this->dbc->query(" + DELETE FROM {$this->transTable} + WHERE id='{$row['id']}' + "); + if(PEAR::isError($r)) return $r; + */ + } + + /** + * + */ + function downloadMetadata($gunid, $asessid) + { + $ret = $this->xmlrpcCall('archive.downloadMetadataOpen', + array('sessid'=>$asessid, 'gunid'=>$gunid) + ); + if(PEAR::isError($ret)) return $ret; + #echo "{$ret['url']}\n"; + if(($mdata = file_get_contents($ret['url'])) === FALSE){ + return PEAR::raiseError("Transport::downloadCronInit: ". + "metadata download failed ({$gunid})", TRERR_MD + ); + } + $filename = $ret['filename']; + $ret = $this->xmlrpcCall('archive.downloadMetadataClose', + array('token'=>$ret['token']) + ); + if(PEAR::isError($ret)) return $ret; + return array('mdata'=>$mdata, 'filename'=>$filename); + } + /* ======================================================= search methods */ /** * Start search in archive @@ -87,259 +765,6 @@ class Transport{ // not implemented yet } - /* ======================================= general file transport methods */ - - /** - * - */ - function uploadOpen($file, $type, $sessid='', $gunid='X') - { - $trid = $this->_createTrId(); - $md5h = $this->_md5sum($this->transDir."/$file"); - $id = $this->dbc->nextId("{$this->transTable}_id_seq"); - $res = $this->dbc->query(" - INSERT INTO {$this->transTable} - (id, trid, direction, state, type, - md5h, url, fname, gunid - ) - VALUES - ($id, '$trid', 'up', 'init', '$type', - '$md5h', '', '$file', '$gunid' - ) - "); - if(PEAR::isError($res)) return $res; -#?? $this->uploadCron(); - return $trid; - } - - /** - * - */ - function uploadCron() - { - // fetch all opened uploads - $rows = $this->dbc->getAll(" - SELECT * FROM {$this->transTable} - WHERE direction='up' AND state<>'closed' - "); - if(count($rows)==0) return TRUE; - $asessid = $this->loginToArchive(); - chdir($this->config['transDir']); - // for all opened uploads: - foreach($rows as $i=>$row){ - switch($row['state']){ - case"init": // ------ new uploads - $finf = $this->xmlrpcCall( 'archive.uploadOpen', - array('sessid'=>$asessid, 'trid'=>$row['trid'], - 'type'=>$row['type'] - ) - ); - if(PEAR::isError($finf)) return $finf; - $res = $this->dbc->query(" - UPDATE {$this->transTable} - SET state='pending', url='{$finf['url']}' - WHERE id='{$row['id']}' - "); - if(PEAR::isError($res)) return $res; - $row['url'] = $finf['url']; -#break; - case"pending": // ------ pending uploads - $finf = $this->uploadCheck($asessid, $row['url']); - if(PEAR::isError($finf)) return $finf; - // test filesize - if(intval($finf['size']) < filesize($row['fname'])){ - // not finished - upload next part - $res = system( - "curl -s -C {$finf['size']} --max-time 600". - " --speed-time 20 --speed-limit 500". - " --connect-timeout 20". - " -T {$row['fname']} {$row['url']}", - $status - ); - }else{ - // hmmm - we are finished? OK - continue - $status = 0; - } - if($status == 0 || $status == 18){ - $finf = $this->uploadCheck($asessid, $row['url']); - if(PEAR::isError($finf)) return $finf; - // test checksum - if($finf['md5h'] == $row['md5h']){ - // finished - $res = $this->dbc->query(" - UPDATE {$this->transTable} SET state='finished' - WHERE id='{$row['id']}' - "); - if(PEAR::isError($res)) return $res; - }else{ - if(intval($finf['size']) >= filesize($row['fname'])) - { - // wrong md5 at finish - TODO: start again - // $this->xmlrpcCall('archive.uploadReset', array()); - return PEAR::raiseError("Transport::uploadCron:". - " file uploaded with bad md5" - ); - } - } - } -#break; - case"finished": // ------ finished uploads - $res = $this->xmlrpcCall( - 'archive.uploadClose', - array('sessid'=>$asessid, - 'url'=>$row['url'], 'type'=>$row['type'], - 'gunid'=>$row['gunid'], - ) - ); - if(PEAR::isError($res)) return $res; - @unlink($this->transDir."/".$row['fname']); - // close upload in db TODO: or delete record? - $this->dbc->query(" - UPDATE {$this->transTable} SET state='closed' - WHERE id='{$row['id']}' - "); - break; - default: - echo "Transport::uploadCron: unknown state". - " '{$row['state']}' (id={$row['id']})\n"; - } // switch state - } // foreach opened - $this->logoutFromArchive($asessid); - return TRUE; - } - - /** - * Check state of uploaded file - * - * @param sessid - * @param url - * @return hash: md5h, size, url - */ - function uploadCheck($sessid, $url) - { - $finf = $this->xmlrpcCall( - 'archive.uploadCheck', - array('sessid'=>$sessid, 'url'=>$url) - ); - return $finf; - } - - /** - * - */ - function downloadOpen($sessid, $type, $gunid, $uid) - { - // insert transport record to db - $trid = $this->_createTrId(); - $id = $this->dbc->nextId("{$this->transTable}_id_seq"); - $res = $this->dbc->query(" - INSERT INTO {$this->transTable} - (id, trid, direction, state, type, - gunid, sessid, uid - ) - VALUES - ($id, '$trid', 'down', 'init', '$type', - '$gunid', '$sessid', $uid - ) - "); - if(PEAR::isError($res)) return $res; -#?? $this->downloadCron(); - return $trid; - } - - /** - * - */ - function downloadCron(&$gb) - { - // fetch all opened downloads - $rows = $this->dbc->getAll(" - SELECT * FROM {$this->transTable} - WHERE direction='down' AND state<>'closed' - "); - if(count($rows)==0) return TRUE; - $asessid = $this->loginToArchive(); - chdir($this->config['transDir']); - // for all opened downloads: - foreach($rows as $i=>$row){ - switch($row['state']){ - case"init": // ------ new downloads - // call archive.downloadOpen - $finf = $this->xmlrpcCall( - 'archive.downloadOpen', - array('sessid'=>$asessid, 'type'=>$row['type'], - 'par'=>$row['gunid'] - ) - ); - if(PEAR::isError($finf)) return $finf; - $res = $this->dbc->query(" - UPDATE {$this->transTable} - SET state='pending', url='{$finf['url']}', - md5h='{$finf['md5h']}', fname='{$finf['fname']}' - WHERE id='{$row['id']}' - "); - if(PEAR::isError($res)) return $res; - $row = array_merge($row, $finf); -#break; - case"pending": // ------ pending downloads - // wget the file - $res = system( - "wget -q -c --timeout={$this->timeout}". - " --waitretry={$this->waitretry}". - " -t {$this->retries} {$row['url']}", - $status - ); - // check consistency - $md5h = $this->_md5sum($row['fname']); - if($status == 0){ - if($md5h == $row['md5h']){ - // mark download as finished - $res = $this->dbc->query(" - UPDATE {$this->transTable} - SET state='finished' - WHERE id='{$row['id']}' - "); - if(PEAR::isError($res)) return $res; - }else{ - @unlink($row['fname']); - } - } -#break; - case"finished": // ------ finished downloads - // call archive that downloads have been finished OK - $res = $this->xmlrpcCall( - 'archive.downloadClose', - array('sessid'=>$asessid, 'url'=>$row['url']) - ); - if(PEAR::isError($res)) return $res; - // process file in fake session - $lsessid = $gb->_fakeSession($row['uid']); - if(PEAR::isError($lsessid)) return $lsessid; - $res = $gb->processTransported( - $lsessid, $row['fname'], $row['type'], $row['gunid'] - ); - if(PEAR::isError($res)) return $res; - $res = $gb->logout($lsessid); - if(PEAR::isError($res)) return $res; - - // close download in db TODO: or delete record? - $res = $this->dbc->query(" - UPDATE {$this->transTable} - SET state='closed' - WHERE id='{$row['id']}' - "); - if(PEAR::isError($res)) return $res; - break; - default: - echo "Transport::downloadCron: unknown state". - " '{$row['state']}' (id={$row['id']})\n"; - } // switch state - } // foreach opened - $this->logoutFromArchive($asessid); - return TRUE; - } - - /* =============================================== authentication methods */ /** @@ -380,10 +805,13 @@ class Transport{ /** * */ - function _createTrId() + function _createTrtok() { - return md5(microtime().$_SERVER['SERVER_ADDR'].rand(). - "org.mdlf.livesupport"); + $initString = + microtime().$_SERVER['SERVER_ADDR'].rand()."org.mdlf.livesupport"; + $hash = md5($initString); + $res = substr($hash, 0, 16); + return $res; } /** @@ -396,64 +824,120 @@ class Transport{ return $resDir; } + /** + * Ping to archive server + * + * @return string sessid or error + */ + function pingToArchive() + { + $res = $this->xmlrpcCall( + 'archive.ping', + array( + 'par'=>'testString_'.date('H:i:s') + ) + ); + return $res; + } /** * XMLRPC call to archive */ function xmlrpcCall($method, $pars=array()) { - $xrp = xmlrpc_encoder($pars); - $c = new xmlrpc_client( + $xrp = XML_RPC_encode($pars); + $c = new XML_RPC_Client( "{$this->config['archiveUrlPath']}/". "{$this->config['archiveXMLRPC']}", $this->config['archiveUrlHost'], $this->config['archiveUrlPort'] ); - $f=new xmlrpcmsg($method, array($xrp)); + $f=new XML_RPC_Message($method, array($xrp)); + #echo "\n--\n".$f->serialize()."\n--\n"; $r = $c->send($f); if ($r->faultCode()>0) { return PEAR::raiseError($r->faultString(), $r->faultCode()); }else{ $v = $r->value(); - return xmlrpc_decoder($v); + return XML_RPC_decode($v); } } /** * md5 checksum of local file */ - function _md5sum($fpath) + function _chsum($fpath) { - $md5h = `md5sum $fpath`; - $arr = split(' ', $md5h); - return $arr[0]; + return md5_file($fpath); } + /** + * logging wrapper for PEAR error object + * + * @param eo PEAR error object + */ + function trLogPear($eo, $row=NULL) + { + $msg = $eo->getMessage()." ".$eo->getUserInfo(); + if(!is_null($row)) $msg .= "\n ".serialize($row); + $this->trLog($msg); + } + + /** + * logging for debug transports + * + * @param msg string - log message + */ + function trLog($msg) + { + $fp=fopen("{$this->transDir}/log", "a") or die("Can't write to log\n"); + fputs($fp, "---".date("H:i:s")."---\n $msg\n"); + fclose($fp); + } + /* ====================================================== install methods */ + /** + * Delete all transports + */ + function resetData() + { + return $this->dbc->query("DELETE FROM {$this->transTable}"); + } + /** * Install method
- * state: pending, finished, closed + * + * direction: up | down + * state: init | pending | finished | closed | failed + * trtype: audioclip | playlist | searchjob + * */ function install() { $this->dbc->query("CREATE TABLE {$this->transTable} ( id int not null, - trid char(32) not null, - direction varchar(128) not null, -- down | up + trtok char(16) not null, -- transport token + direction varchar(128) not null, state varchar(128) not null, - type varchar(128) not null, -- file | searchJob - md5h char(32), + trtype varchar(128) not null, + gunid bigint, -- global unique id + pdtoken bigint, -- put/download token from archive url varchar(255), - fname varchar(255), - gunid char(32), - sessid char(32), - uid int, - parid int, + fname varchar(255), -- mnemonic filename + localfile varchar(255), -- pathname of local part + expectedsum char(32), -- expected file checksum + realsum char(32), -- checksum of transported part + expectedsize int, -- expected filesize in bytes + realsize int, -- filesize of transported part + uid int, -- local user id of transport owner + parid int, -- local id of download destination folder ts timestamp )"); $this->dbc->createSequence("{$this->transTable}_id_seq"); $this->dbc->query("CREATE UNIQUE INDEX {$this->transTable}_id_idx ON {$this->transTable} (id)"); - $this->dbc->query("CREATE INDEX {$this->transTable}_trid_idx - ON {$this->transTable} (trid)"); + $this->dbc->query("CREATE UNIQUE INDEX {$this->transTable}_trtok_idx + ON {$this->transTable} (trtok)"); + $this->dbc->query("CREATE UNIQUE INDEX {$this->transTable}_token_idx + ON {$this->transTable} (token)"); $this->dbc->query("CREATE INDEX {$this->transTable}_gunid_idx ON {$this->transTable} (gunid)"); } diff --git a/livesupport/modules/storageServer/var/cron/transportCron.php b/livesupport/modules/storageServer/var/cron/transportCron.php index 2cfab1ed5..57679657a 100755 --- a/livesupport/modules/storageServer/var/cron/transportCron.php +++ b/livesupport/modules/storageServer/var/cron/transportCron.php @@ -1,5 +1,6 @@ #!/usr/bin/php -q setFetchMode(DB_FETCHMODE_ASSOC); $gb = &new LocStor(&$dbc, $config); +$tr =& new Transport(&$dbc, &$gb, $config); +$cnt = 1; -$res = $gb->cronJob(); +#$res = $gb->cronJob(); +#var_dump($res); -var_dump($res); +for($i=0; $i<$cnt; $i++){ + $r = $tr->uploadCron(); + if(!$r) exit(1); +} + +for($i=0; $i<$cnt; $i++){ + $r = $tr->downloadCron(); + if(!$r) exit(1); +} + +exit(0); ?> \ No newline at end of file diff --git a/livesupport/modules/storageServer/var/xmlrpc/XR_LocStor.php b/livesupport/modules/storageServer/var/xmlrpc/XR_LocStor.php index e93047dff..84e11e766 100644 --- a/livesupport/modules/storageServer/var/xmlrpc/XR_LocStor.php +++ b/livesupport/modules/storageServer/var/xmlrpc/XR_LocStor.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.8 $ + Version : $Revision: 1.9 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/xmlrpc/XR_LocStor.php,v $ ------------------------------------------------------------------------------*/ @@ -188,7 +188,7 @@ class XR_LocStor extends LocStor{ *
  • sessid : string - session id
    • *
  • gunid : string - global unique id of AudioCLip
    • *
  • metadata : string - metadata XML string
    • - *
  • fname : string - human readable menmonic file name + *
  • fname : string - human readable mnemonic file name * with extension corresponding to filetype
    • *
  • chsum : string - md5 checksum of media file
    • * @@ -254,6 +254,7 @@ class XR_LocStor extends LocStor{ *
  • 801 - wrong 1st parameter, struct expected.
    • *
  • 805 - xr_storeAudioClipClose: * <message from lower layer>
    • + *
  • 850 - wrong 1st parameter, struct expected.
    • * * * @param input XMLRPC struct @@ -266,7 +267,9 @@ class XR_LocStor extends LocStor{ if(!$ok) return $r; $res = $this->storeAudioClipClose($r['sessid'], $r['token']); if(PEAR::isError($res)){ - return new XML_RPC_Response(0, 805, + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_TOKEN ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_storeAudioClipClose: ".$res->getMessage(). " ".$res->getUserInfo() ); @@ -384,6 +387,9 @@ class XR_LocStor extends LocStor{ * * * On errors, returns an XML-RPC error response. @@ -394,6 +400,7 @@ class XR_LocStor extends LocStor{ *
  • 801 - wrong 1st parameter, struct expected.
    • *
  • 805 - xr_accessRawAudioDataOpen: * <message from lower layer>
    • + *
  • 847 - invalid gunid.
    • * * * @param input XMLRPC struct @@ -406,7 +413,9 @@ class XR_LocStor extends LocStor{ if(!$ok) return $r; $res = $this->downloadRawAudioDataOpen($r['sessid'], $r['gunid']); if(PEAR::isError($res)){ - return new XML_RPC_Response(0, 805, + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_NOTF ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_downloadRawAudioDataOpen: ".$res->getMessage(). " ".$res->getUserInfo() ); @@ -477,6 +486,8 @@ class XR_LocStor extends LocStor{ * * * On errors, returns an XML-RPC error response. @@ -497,6 +508,7 @@ class XR_LocStor extends LocStor{ { list($ok, $r) = $this->_xr_getPars($input); if(!$ok) return $r; + #$this->debugLog("{$r['sessid']}, {$r['gunid']}"); $res = $this->downloadMetadataOpen($r['sessid'], $r['gunid']); if(PEAR::isError($res)){ return new XML_RPC_Response(0, 805, @@ -663,6 +675,7 @@ class XR_LocStor extends LocStor{ * * * On errors, returns an XML-RPC error response. @@ -802,6 +815,7 @@ class XR_LocStor extends LocStor{ * * * On errors, returns an XML-RPC error response. @@ -812,6 +826,7 @@ class XR_LocStor extends LocStor{ *
  • 801 - wrong 1st parameter, struct expected.
    • *
  • 805 - xr_accessPlaylist: * <message from lower layer>
    • + *
  • 847 - invalid plid.
    • * * * @param input XMLRPC struct @@ -824,7 +839,9 @@ class XR_LocStor extends LocStor{ if(!$ok) return $r; $res = $this->accessPlaylist($r['sessid'], $r['plid']); if(PEAR::isError($res)){ - return new XML_RPC_Response(0, 805, + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_NOTF ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_accessPlaylist: ".$res->getMessage(). " ".$res->getUserInfo() ); @@ -970,7 +987,7 @@ class XR_LocStor extends LocStor{ return new XML_RPC_Response(XML_RPC_encode(array('available'=>$res))); } - /* ----------------------------------------------------------------- etc. */ + /* --------------------------------------------------------- info methods */ /** * Check if audio clip exists and return TRUE/FALSE * @@ -1017,6 +1034,7 @@ class XR_LocStor extends LocStor{ return new XML_RPC_Response(XML_RPC_encode(array('exists'=>$res))); } + /* ----------------------------------------------------- metadata methods */ /** * Return all file's metadata as XML string * @@ -1181,49 +1199,7 @@ class XR_LocStor extends LocStor{ return new XML_RPC_Response(XML_RPC_encode($res)); } - /** - * Reset storageServer for debugging. - * - * The XML-RPC name of this method is "locstor.resetStorage". - * - * The input parameters are an empty XML-RPC struct. - * - * On success, returns a XML-RPC struct with following - * fields: - * - * - * On errors, returns an XML-RPC error response. - * The possible error codes and error message are: - * - * - * @param input XMLRPC struct - * @return XMLRPC struct - * @see LocStor::getAudioClip - */ - function xr_resetStorage($input) - { - list($ok, $r) = $this->_xr_getPars($input); - if(!$ok) return $r; - $res = $this->resetStorage(); - if(PEAR::isError($res)){ - return new XML_RPC_Response(0, 805, - "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() - ); - } - return new XML_RPC_Response(XML_RPC_encode($res)); - } - + /* ---------------------------------------------- methods for preferences */ /** * Load user preference value * @@ -1265,8 +1241,9 @@ class XR_LocStor extends LocStor{ $pr =& new Prefs(&$this); $res = $pr->loadPref($r['sessid'], $r['key']); if(PEAR::isError($res)){ - $ec = intval($res->getCode()); - return new XML_RPC_Response(0, 800+($ec == 48 || $ec == 49 ? $ec : 5 ), + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS || $ec0 == GBERR_PREF ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() ); @@ -1315,9 +1292,9 @@ class XR_LocStor extends LocStor{ $pr =& new Prefs(&$this); $res = $pr->savePref($r['sessid'], $r['key'], $r['value']); if(PEAR::isError($res)){ - #return new XML_RPC_Response(0, 805, - $ec = intval($res->getCode()); - return new XML_RPC_Response(0, 800+($ec == 48 ? $ec : 5 ), + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() ); } @@ -1365,16 +1342,214 @@ class XR_LocStor extends LocStor{ $pr =& new Prefs(&$this); $res = $pr->delPref($r['sessid'], $r['key']); if(PEAR::isError($res)){ - #return new XML_RPC_Response(0, 805, - $ec = intval($res->getCode()); - return new XML_RPC_Response(0, 800+($ec == 48 || $ec == 49 ? $ec : 5 ), + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS || $ec0 == GBERR_PREF ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() ); } return new XML_RPC_Response(XML_RPC_encode(array('status'=>$res))); } - /* ------------------------------------------- test methods for debugging */ + /* -------------------------------------------- remote repository methods */ + /** + * Starts upload audioclip to remote archive + * + * The XML-RPC name of this method is "locstor.uploadToArchive". + * + * The input parameters are an XML-RPC struct with the following + * fields: + * + * + * On success, returns a XML-RPC struct with single field: + * + * + * On errors, returns an XML-RPC error response. + * The possible error codes and error message are: + * + * + * @param input XMLRPC struct + * @return XMLRPC struct + * @see Pref::uploadToArchive + */ + function xr_uploadToArchive($input) + { + list($ok, $r) = $this->_xr_getPars($input); + if(!$ok) return $r; + require_once '../../../storageServer/var/Transport.php'; + $tr =& new Transport(&$this->dbc, &$this, $this->config); + $res = $tr->uploadToArchive($r['gunid'], $r['sessid']); + if(PEAR::isError($res)){ + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, + "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() + ); + } + return new XML_RPC_Response(XML_RPC_encode(array('trtok'=>$res))); + } + + /** + * Starts download audioclip from remote archive + * + * The XML-RPC name of this method is "locstor.downloadFromArchive". + * + * The input parameters are an XML-RPC struct with the following + * fields: + * + * + * On success, returns a XML-RPC struct with single field: + * + * + * On errors, returns an XML-RPC error response. + * The possible error codes and error message are: + * + * + * @param input XMLRPC struct + * @return XMLRPC struct + * @see Pref::downloadFromArchive + */ + function xr_downloadFromArchive($input) + { + list($ok, $r) = $this->_xr_getPars($input); + if(!$ok) return $r; + require_once '../../../storageServer/var/Transport.php'; + $tr =& new Transport(&$this->dbc, &$this, $this->config); + $res = $tr->downloadFromArchive($r['gunid'], $r['sessid']); + if(PEAR::isError($res)){ + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, + "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() + ); + } + return new XML_RPC_Response(XML_RPC_encode(array('trtok'=>$res))); + } + + /** + * Checking status of transported file + * + * The XML-RPC name of this method is "locstor.getTransportInfo". + * + * The input parameters are an XML-RPC struct with the following + * fields: + * + * + * On success, returns a XML-RPC struct with the following fields: + * + * + * On errors, returns an XML-RPC error response. + * The possible error codes and error message are: + * + * + * @param input XMLRPC struct + * @return XMLRPC struct + * @see Pref::getTransportInfo + */ + function xr_getTransportInfo($input) + { + list($ok, $r) = $this->_xr_getPars($input); + if(!$ok) return $r; + require_once '../../../storageServer/var/Transport.php'; + $tr =& new Transport(&$this->dbc, &$this, $this->config); + $res = $tr->getTransportInfo($r['trtok'], $r['sessid']); + if(PEAR::isError($res)){ + $ec0 = intval($res->getCode()); + $ec = ($ec0 == GBERR_SESS || $ec0 == TRERR_TOK ? 800+$ec0 : 805 ); + return new XML_RPC_Response(0, $ec, + "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() + ); + } + return new XML_RPC_Response(XML_RPC_encode($res)); + } + + /* ------------------------------------------------ methods for debugging */ + /** + * Reset storageServer for debugging. + * + * The XML-RPC name of this method is "locstor.resetStorage". + * + * The input parameters are an empty XML-RPC struct. + * + * On success, returns a XML-RPC struct with following + * fields: + * + * + * On errors, returns an XML-RPC error response. + * The possible error codes and error message are: + * + * + * @param input XMLRPC struct + * @return XMLRPC struct + * @see LocStor::getAudioClip + */ + function xr_resetStorage($input) + { + list($ok, $r) = $this->_xr_getPars($input); + if(!$ok) return $r; + $res = $this->resetStorage(); + if(PEAR::isError($res)){ + return new XML_RPC_Response(0, 805, + "xr_getAudioClip: ".$res->getMessage()." ".$res->getUserInfo() + ); + } + return new XML_RPC_Response(XML_RPC_encode($res)); + } + /** * Test XMLRPC - strupper and return given string, * also return loginname of logged user diff --git a/livesupport/modules/storageServer/var/xmlrpc/xrLocStor.php b/livesupport/modules/storageServer/var/xmlrpc/xrLocStor.php index 2c9300450..9af8e1f1f 100644 --- a/livesupport/modules/storageServer/var/xmlrpc/xrLocStor.php +++ b/livesupport/modules/storageServer/var/xmlrpc/xrLocStor.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.17 $ + Version : $Revision: 1.18 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/xmlrpc/xrLocStor.php,v $ ------------------------------------------------------------------------------*/ @@ -117,6 +117,10 @@ $methods = array( 'loadPref' => 'Load user preference value.', 'savePref' => 'Save user preference value.', 'delPref' => 'Delete user preference record.', + + 'uploadToArchive' => 'Starts upload audioclip to remote archive.', + 'downloadFromArchive' => 'Starts download audioclip from remote archive.', + 'getTransportInfo' => 'Checking status of transported file.', ); $defs = array(); diff --git a/livesupport/modules/storageServer/var/xmlrpc/xr_cli_test.php b/livesupport/modules/storageServer/var/xmlrpc/xr_cli_test.php index 13dcf4fe3..621018482 100644 --- a/livesupport/modules/storageServer/var/xmlrpc/xr_cli_test.php +++ b/livesupport/modules/storageServer/var/xmlrpc/xr_cli_test.php @@ -23,7 +23,7 @@ Author : $Author: tomas $ - Version : $Revision: 1.1 $ + Version : $Revision: 1.2 $ Location : $Source: /home/paul/cvs2svn-livesupport/newcvsrepo/livesupport/modules/storageServer/var/xmlrpc/xr_cli_test.php,v $ ------------------------------------------------------------------------------*/ @@ -122,6 +122,15 @@ $infos = array( 'p'=>array('sessid', 'key', 'value'), 'r'=>'status'), "delPref" => array('m'=>"locstor.delPref", 'p'=>array('sessid', 'key'), 'r'=>'status'), + + "uploadToArchive" => array('m'=>"locstor.uploadToArchive", + 'p'=>array('sessid', 'gunid'), 'r'=>'trtok'), + "downloadFromArchive" => array('m'=>"locstor.downloadFromArchive", + 'p'=>array('sessid', 'gunid'), 'r'=>'trtok'), + "getTransportInfo" => array('m'=>"locstor.getTransportInfo", + 'p'=>array('sessid', 'trtok'), + 'r'=>array('state', 'realsize', 'realsum', 'expectedsize', 'expectedsum')), + "openPut" => array('m'=>"locstor.openPut", 'p'=>array()), "closePut" => array('m'=>"locstor.closePut", 'p'=>array()), );