<?php
/**
*
* @author Benjamin Gillissen <hide@address.com>
*
* **************************************************************
Copyright (C) 2009 Benjamin Gillissen
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; either version 2
of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details at:
http://www.gnu.org/copyleft/gpl.html
* **************************************************************
*/
class hellarpcs {
private static $_RDB;
private static $_CACHE;
private static $_NET;
private static $_CPOL;
private static function init(){
//conf/status backend
if ( isset(self::$_RDB) ){ return; }
if ( configs::get('hellarpc', 'mode') == 'db' ){
self::$_RDB = new rpconf_db();
} else {
self::$_RDB = new rpconf_array();
}
$ct = configs::get('hellarpc', 'cachetype');
$cc = configs::get('hellarpc', 'cacheconf');
$c = "rpc_cache_$ct";
self::$_CACHE = new $c($cc);
}
public static function load($rid, $cpol=NULL){
self::init();
//cache policy....
$dbpol = self::$_RDB->getpolicy($rid);
if ( $cpol === NULL AND $dbpol !== FALSE ){
self::$_CPOL[$rid] = $dbpol;
} else {
self::$_CPOL[$rid] = $cpol;
}
unset($cpol,$dbpol);
//net rpc
if ( isset(self::$_NET[$rid]) ){ return TRUE; }
$conf = self::$_RDB->getconninfo($rid);
self::$_NET[$rid] = new hellarpc_net($rid, $conf);
return TRUE;
}
private static function loadcheck($rid){
if ( !isset(self::$_NET[$rid]) ){
errors::raise('You must load a rpc before using it.', CORE_LOG_ERROR, 'RPC');
return FALSE;
}
}
public static function getstatus($rid){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if (self::$_CPOL[$rid] == -1 ){
if ( self::$_CACHE->status_isexpired($rid, self::$_RDB->get_staTO($rid)) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
} else {
$r = self::$_CACHE->get_status($rid);
}
} elseif ( self::$_CPOL[$rid] == 1 ){
$r = self::$_CACHE->get_status($rid);
} elseif ( self::$_CPOL[$rid] == 0 ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
}
return $r;
}
public static function isrunning($rid){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
return is_array(self::getstatus($rid));
}
public static function ispaused($rid){
if ( FALSE === self::isrunning($rid) ){ return FALSE; }
$r = self::getstatus($rid);
return $r['is_paused'];
}
public static function got_qitems($rid){
if ( FALSE === self::isrunning($rid) ){ return FALSE; }
$r = self::getstatus($rid);
return is_array($r['queued']);
}
public static function isdloading($rid, $did=NULL){
if ( FALSE === self::isrunning($rid) ){ return FALSE; }
$r = self::getstatus($rid);
if ( ! is_array(@$r['currently_downloading']) ){ return FALSE; }
if ( $did === NULL ){ return ( count($r['currently_downloading']) != 0 ); }
foreach($r['currently_downloading'] as $k => $item ){
if ( @$item['msgid'] == $did ){ return TRUE; }
}
return FALSE;
}
public static function isprocessing($rid, $did=NULL){
if ( FALSE === self::isrunning($rid) ){ return FALSE; }
$r = self::getstatus($rid);
if ( ! is_array(@$r['currently_processing']) ){ return FALSE; }
if ( $did === NULL ){ return ( count($r['currently_processing']) != 0 ); }
foreach($r['currently_processing'] as $k => &$item ){
if ( @$item['msgid'] == $did ){ return TRUE; }
}
return FALSE;
}
public static function isqueued($rid, $did){
if ( FALSE === self::isrunning($rid) ){ return FALSE; }
$r = self::getstatus($rid);
if ( ! is_array(@$r['queued']) ){ return FALSE; }
foreach($r['queued'] as $k => &$item ){
if ( @$item['msgid'] == $did ){ return TRUE; }
}
return FALSE;
}
//DIRECT RPC CALL, need active_tcp_conn
public static function cancel($rid, $did=NULL){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
If ( FALSE !== self::$_NET[$rid]->cancel($did) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function enqueue($rid, $did){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->enqueue($did) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function movedown($rid, $id){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->movedown($id) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function moveup($rid, $id){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->moveup($id) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function movetop($rid, $id){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->movetop($id) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function movebot($rid, $id){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->movebot($id) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function force($rid, $id){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->force($id) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function pause($rid){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
If ( FALSE !== self::$_NET[$rid]->pause() ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function resume($rid){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->resume() ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
public static function setrate($rid, $rate){
if ( FALSE === self::loadcheck($rid) ){ return FALSE; }
if ( FALSE == self::isactif($rid) ){ return FALSE; }
if ( FALSE !== self::$_NET[$rid]->setrate($rate) ){
$r = self::$_NET[$rid]->getstatus();
self::$_CACHE->set_status($rid, $r);
return TRUE;
}
return FALSE;
}
//FORWARD TO RPC CONF ENGINE
public static function isrpc($rid){ self::init();return self::$_RDB->isrpc($rid); }
public static function isactif($rid){ self::init();return self::$_RDB->isactif($rid); }
public static function getname($rid){ self::init();return self::$_RDB->getname($rid); }
public static function getpolicy($rid){ self::init();return self::$_RDB->getpolicy($rid); }
public static function get_host($rid){ self::init();return self::$_RDB->gethost($rid); }
public static function get_destrepo($rid){ self::init();return self::$_RDB->get_destrepo($rid); }
public static function get_repobaseurl($rid){ self::init();return self::$_RDB->get_repobaseurl($rid); }
public static function getfirst(){ self::init();return self::$_RDB->getfirst();}
public static function listall($start=NULL, $end=NULL){ self::init();return self::$_RDB->listall($start, $end); }
public static function countall(){ self::init();return self::$_RDB->countall(); }
//DID explode-implode
public static function gendid($type, $id){
// return $type.':'.$id;
switch($type){
case 'ftrid' : $id = "0".$id;
break;
case 'nzbid' : $id = "00".$id;
break;
}
return $id;
}
public static function splitdid($did){
//list($ar['type'], $ar['id']) = split(':', $did);
if ( TRUE == ereg("^00", $did) ){
$ar['type'] = 'nzbid';
$ar['id'] = ereg_replace("^00", '', $did);
} elseif ( TRUE == ereg("^0", $did) ){
$ar['type'] = 'ftrid';
$ar['id'] = ereg_replace("^0", '', $did);
} elseif ( FALSE === ereg("^0", $did) ){
$ar['type'] = 'nbrid';
$ar['id'] = $did;
} else {
echo 'nomatch';
return FALSE;
}
return $ar;
}
//FORWARD TO RPC CACHE
public static function get_lastreq($rpc){ self::init();return self::$_CACHE->get_lastreq($rpc); }
public static function get_lastbrw($rpc){ self::init();return self::$_CACHE->get_lastbrw($rpc); }
public static function set_lastbrw($rpc, $time){ self::init();return self::$_CACHE->set_lastbrw($rpc, $time); }
public static function clean_browser($rpc){ self::init();return self::$_CACHE->clean_browser($rpc); }
public static function hash_iscached($rpc, $catid, $hash){ self::init();return self::$_CACHE->hash_iscached($rpc, $catid, $hash); }
public static function insert_item(&$ITEM){ self::init();return self::$_CACHE->insert_item(&$ITEM); }
public static function update_item(&$ITEM){ self::init();return self::$_CACHE->update_item(&$ITEM); }
public static function listbycat($rpc, $catid, $order, $way){ self::init();return self::$_CACHE->listbycat($rpc, $catid, $order, $way); }
//search related
public static function idisdloading($rpc, $engine, $id){ self::load($rpc);return self::isdloading($rpc, self::gendid($engine, $id)); }
public static function idisprocessing($rpc, $engine, $id){ self::load($rpc);return self::isprocessing($rpc, self::gendid($engine, $id)); }
public static function idisqueued($rpc, $engine, $id){ self::load($rpc);return self::isqueued($rpc, self::gendid($engine, $id)); }
public static function idisdelayed($rpc, $engine, $id){
if ( $engine !== 'nbrid' ){ return FALSE; }
self::init();
return self::$_CACHE->id_isdelayed($rpc, $id);
}
public static function idisonrpc($rpc, $engine, $id){ self::init();return self::$_CACHE->idisonrpc($rpc, $engine, $id); }
//newzbin DIDrate related, use cache engine for storage
public static function gotfree_slot(){ self::init();return self::$_CACHE->gotfree_slot(); }
public static function lock_slot($reason, $id){ self::init();return self::$_CACHE->lock_slot($reason, $id); }
public static function list_slot(){
self::init();
$slots = self::$_CACHE->list_slot();
if ( FALSE === $slots ){ return FALSE; }
foreach($slots as $k => &$slot){
$slot['freein'] = 60 - ( time() - $slot['date'] );
}
return $slots;
}
public static function delayed_enqueue($rpc, $id){ self::init();return self::$_CACHE->delayed_enqueue($rpc, $id); }
public static function cancel_delayed($rpc, $unic){ self::init();return self::$_CACHE->cancel_delayed($rpc, $unic); }
public static function list_delayed($rpc=NULL){ self::init();return self::$_CACHE->list_delayed($rpc);}
public static function enqueue_delayed(){
self::init();
$e=Array();
$r=0;
$f=0;
while( TRUE ){
$item = self::$_CACHE->delay_getoldest();
if ( FALSE === $item ){ break; }
if ( FALSE === self::lock_slot('Delay_Enqueue', $item['id']) ){ break; }
if ( FALSE !== self::enqueue($item['rpc'], $item['id']) ){
self::$_CACHE->cancel_delayed($item['rpc'], $item['unic']);
if ( !isset($e[$item['rpc']])) { $e[$item['rpc']] = 0;$r++; }
$e[$item['rpc']]++;
}
}
if ( $r != 0 ){
$t=0;
foreach($e as $rpc => &$c){ $t += $c; }
pagegen::add_event("$t Delayed item(s) enqueued on $r RPC(s), $f failed.");
}
}
public static function delete_delayed(){ }
}