From 3c220dd40cc5cabb991d23f20636f2a973d6806c Mon Sep 17 00:00:00 2001 From: Michael Date: Mon, 27 Feb 2017 21:07:32 +0000 Subject: [PATCH] Queue.php is now reworked --- include/queue.php | 235 ++++++++++++++++++++-------------------------- 1 file changed, 103 insertions(+), 132 deletions(-) diff --git a/include/queue.php b/include/queue.php index 6bbfdaae81..509fb19d26 100644 --- a/include/queue.php +++ b/include/queue.php @@ -1,7 +1,5 @@ 1) + if ($argc > 1) $queue_id = intval($argv[1]); else $queue_id = 0; - $deadguys = array(); - $deadservers = array(); - $serverlist = array(); - if (!$queue_id) { logger('queue: start'); @@ -34,7 +28,7 @@ function queue_run(&$argv, &$argc){ if ($r) { foreach ($r as $rr) { logger('queue: deliverq'); - proc_run(PRIORITY_HIGH,'include/delivery.php',$rr['cmd'],$rr['item'],$rr['contact']); + proc_run(PRIORITY_HIGH,'include/delivery.php', $rr['cmd'], $rr['item'], $rr['contact']); } } @@ -53,142 +47,119 @@ function queue_run(&$argv, &$argc){ // After that, we'll only attempt delivery once per hour. $r = q("SELECT `id` FROM `queue` WHERE ((`created` > UTC_TIMESTAMP() - INTERVAL 12 HOUR && `last` < UTC_TIMESTAMP() - INTERVAL 15 MINUTE) OR (`last` < UTC_TIMESTAMP() - INTERVAL 1 HOUR)) ORDER BY `cid`, `created`"); - } else { - logger('queue: start for id '.$queue_id); - $r = q("SELECT `id` FROM `queue` WHERE `id` = %d LIMIT 1", - intval($queue_id) - ); - } + call_hooks('queue_predeliver', $a, $r); - if (!$r){ + if (dbm::is_result) { + foreach ($r as $q_item) { + logger('Call queue for id '.$q_item['id']); + proc_run(PRIORITY_LOW, "include/queue.php", $q_item['id']); + } + } return; } - if (!$queue_id) - call_hooks('queue_predeliver', $a, $r); - - // delivery loop + // delivering require_once('include/salmon.php'); require_once('include/diaspora.php'); - foreach($r as $q_item) { + $r = q("SELECT * FROM `queue` WHERE `id` = %d LIMIT 1", + intval($queue_id)); - // queue_predeliver hooks may have changed the queue db details, - // so check again if this entry still needs processing - - if ($queue_id) { - $qi = q("SELECT * FROM `queue` WHERE `id` = %d LIMIT 1", - intval($queue_id)); - } else { - logger('Call queue for id '.$q_item['id']); - proc_run(PRIORITY_LOW, "include/queue.php", $q_item['id']); - continue; - } - - if(! count($qi)) - continue; - - - $c = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1", - intval($qi[0]['cid']) - ); - if (! dbm::is_result($c)) { - remove_queue_item($q_item['id']); - continue; - } - if(in_array($c[0]['notify'],$deadguys)) { - logger('queue: skipping known dead url: ' . $c[0]['notify']); - update_queue_time($q_item['id']); - continue; - } - - $server = poco_detect_server($c[0]['url']); - - if (($server != "") AND !in_array($server, $serverlist)) { - logger("Check server ".$server." (".$c[0]["network"].")"); - if (!poco_check_server($server, $c[0]["network"], true)) - $deadservers[] = $server; - - $serverlist[] = $server; - } - - if (($server != "") AND in_array($server, $deadservers)) { - logger('queue: skipping known dead server: '.$server); - update_queue_time($q_item['id']); - continue; - } - - $u = q("SELECT `user`.*, `user`.`pubkey` AS `upubkey`, `user`.`prvkey` AS `uprvkey` - FROM `user` WHERE `uid` = %d LIMIT 1", - intval($c[0]['uid']) - ); - if (! dbm::is_result($u)) { - remove_queue_item($q_item['id']); - continue; - } - - $data = $qi[0]['content']; - $public = $qi[0]['batch']; - $contact = $c[0]; - $owner = $u[0]; - - $deliver_status = 0; - - switch($contact['network']) { - case NETWORK_DFRN: - logger('queue: dfrndelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); - $deliver_status = dfrn::deliver($owner,$contact,$data); - - if($deliver_status == (-1)) { - update_queue_time($q_item['id']); - $deadguys[] = $contact['notify']; - } else - remove_queue_item($q_item['id']); - - break; - case NETWORK_OSTATUS: - if($contact['notify']) { - logger('queue: slapdelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); - $deliver_status = slapper($owner,$contact['notify'],$data); - - if($deliver_status == (-1)) { - update_queue_time($q_item['id']); - $deadguys[] = $contact['notify']; - } else - remove_queue_item($q_item['id']); - } - break; - case NETWORK_DIASPORA: - if($contact['notify']) { - logger('queue: diaspora_delivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); - $deliver_status = Diaspora::transmit($owner,$contact,$data,$public,true); - - if($deliver_status == (-1)) { - update_queue_time($q_item['id']); - $deadguys[] = $contact['notify']; - } else - remove_queue_item($q_item['id']); - - } - break; - - default: - $params = array('owner' => $owner, 'contact' => $contact, 'queue' => $q_item, 'result' => false); - call_hooks('queue_deliver', $a, $params); - - if($params['result']) - remove_queue_item($q_item['id']); - else - update_queue_time($q_item['id']); - - break; - - } - logger('Deliver status '.$deliver_status.' for item '.$q_item['id'].' to '.$contact['name'].' <'.$contact['url'].'>'); + if (!dbm::is_result($r)) { + return; } + $q_item = $r[0]; + + $c = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1", + intval($q_item['cid']) + ); + + if (!dbm::is_result($c)) { + remove_queue_item($q_item['id']); + return; + } + + $server = poco_detect_server($c[0]['url']); + + if ($server != "") { + logger("Check server ".$server." (".$c[0]["network"].")"); + + if (!poco_check_server($server, $c[0]["network"], true)) { + logger('queue: skipping dead server: '.$server); + update_queue_time($q_item['id']); + return; + } + } + + $u = q("SELECT `user`.*, `user`.`pubkey` AS `upubkey`, `user`.`prvkey` AS `uprvkey` + FROM `user` WHERE `uid` = %d LIMIT 1", + intval($c[0]['uid']) + ); + if (!dbm::is_result($u)) { + remove_queue_item($q_item['id']); + return; + } + + $data = $q_item['content']; + $public = $q_item['batch']; + $contact = $c[0]; + $owner = $u[0]; + + $deliver_status = 0; + + switch ($contact['network']) { + case NETWORK_DFRN: + logger('queue: dfrndelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); + $deliver_status = dfrn::deliver($owner, $contact, $data); + + if ($deliver_status == (-1)) { + update_queue_time($q_item['id']); + } else { + remove_queue_item($q_item['id']); + } + break; + case NETWORK_OSTATUS: + if ($contact['notify']) { + logger('queue: slapdelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); + $deliver_status = slapper($owner, $contact['notify'], $data); + + if ($deliver_status == (-1)) { + update_queue_time($q_item['id']); + } else { + remove_queue_item($q_item['id']); + } + } + break; + case NETWORK_DIASPORA: + if ($contact['notify']) { + logger('queue: diaspora_delivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>'); + $deliver_status = Diaspora::transmit($owner, $contact, $data, $public, true); + + if ($deliver_status == (-1)) { + update_queue_time($q_item['id']); + } else { + remove_queue_item($q_item['id']); + } + } + break; + + default: + $params = array('owner' => $owner, 'contact' => $contact, 'queue' => $q_item, 'result' => false); + call_hooks('queue_deliver', $a, $params); + + if ($params['result']) + remove_queue_item($q_item['id']); + else + update_queue_time($q_item['id']); + + break; + + } + logger('Deliver status '.$deliver_status.' for item '.$q_item['id'].' to '.$contact['name'].' <'.$contact['url'].'>'); + return; }