From 846c4cea7c0e3868a63a187ee9a504a031b2a7e4 Mon Sep 17 00:00:00 2001 From: Friendika Date: Sun, 28 Aug 2011 21:41:42 -0700 Subject: [PATCH] implement delivery queue in case notifier gets killed --- boot.php | 2 +- database.sql | 6 ++++++ include/delivery.php | 6 ++++++ include/diaspora.php | 5 ++--- include/items.php | 4 ++-- include/notifier.php | 45 +++++++++++++++++++++++++++++++------------- include/queue.php | 14 ++++++++++++++ update.php | 12 +++++++++++- 8 files changed, 74 insertions(+), 20 deletions(-) diff --git a/boot.php b/boot.php index 060bd711..595dc1fe 100644 --- a/boot.php +++ b/boot.php @@ -9,7 +9,7 @@ require_once("include/pgettext.php"); define ( 'FRIENDIKA_VERSION', '2.2.1086' ); define ( 'DFRN_PROTOCOL_VERSION', '2.21' ); -define ( 'DB_UPDATE_VERSION', 1083 ); +define ( 'DB_UPDATE_VERSION', 1084 ); define ( 'EOL', "
\r\n" ); define ( 'ATOM_TIME', 'Y-m-d\TH:i:s\Z' ); diff --git a/database.sql b/database.sql index f6ae4c7c..9819914f 100644 --- a/database.sql +++ b/database.sql @@ -606,3 +606,9 @@ INDEX ( `iid` ) ) ENGINE = MyISAM DEFAULT CHARSET=utf8; +CREATE TABLE IF NOT EXISTS `deliverq` ( +`id` INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY , +`cmd` CHAR( 32 ) NOT NULL , +`item` INT NOT NULL , +`contact` INT NOT NULL +) ENGINE = MyISAM DEFAULT CHARSET=utf8; diff --git a/include/delivery.php b/include/delivery.php index 0df8ea7e..1f5883c2 100644 --- a/include/delivery.php +++ b/include/delivery.php @@ -37,6 +37,12 @@ function delivery_run($argv, $argc){ $item_id = intval($argv[2]); $contact_id = intval($argv[3]); + q("delete from deliverq where cmd = '%s' and item = %d and contact = %d limit 1", + dbesc($cmd), + dbesc($item_id), + dbesc($contact_id) + ); + if((! $item_id) || (! $contact_id)) return; diff --git a/include/diaspora.php b/include/diaspora.php index 6cba0ece..99bc21c0 100644 --- a/include/diaspora.php +++ b/include/diaspora.php @@ -140,10 +140,9 @@ EOT; $encrypted_outer_key_bundle = ''; openssl_public_encrypt($outer_json,$encrypted_outer_key_bundle,$pubkey); - logger('outer_bundle_encrypt: ' . openssl_error_string()); $b64_encrypted_outer_key_bundle = base64_encode($encrypted_outer_key_bundle); - logger('outer_bundle: ' . $b64_encrypted_outer_key_bundle . ' key: ' . $pubkey); + logger('outer_bundle: ' . $b64_encrypted_outer_key_bundle . ' key: ' . $pubkey, LOGGER_DATA); $encrypted_header_json_object = json_encode(array('aes_key' => base64_encode($encrypted_outer_key_bundle), 'ciphertext' => base64_encode($ciphertext))); @@ -223,7 +222,7 @@ function diaspora_decode($importer,$xml) { * */ - logger('decrypted: ' . $decrypted); + logger('decrypted: ' . $decrypted, LOGGER_DEBUG); $idom = parse_xml_string($decrypted,false); $inner_iv = base64_decode($idom->iv); diff --git a/include/items.php b/include/items.php index e9594cff..1603dec6 100644 --- a/include/items.php +++ b/include/items.php @@ -927,7 +927,7 @@ function dfrn_deliver($owner,$contact,$atom, $dissolve = false) { if(! $curl_stat) return(-1); // timed out - logger('dfrn_deliver: ' . $xml); + logger('dfrn_deliver: ' . $xml, LOGGER_DATA); if(! $xml) return 3; @@ -991,7 +991,7 @@ function dfrn_deliver($owner,$contact,$atom, $dissolve = false) { $key = substr(random_string(),0,16); $data = bin2hex(aes_encrypt($postvars['data'],$key)); $postvars['data'] = $data; - logger('rino: sent key = ' . $key); + logger('rino: sent key = ' . $key, LOGGER_DEBUG); if($dfrn_version >= 2.1) { diff --git a/include/notifier.php b/include/notifier.php index b87aa95b..cde156cb 100644 --- a/include/notifier.php +++ b/include/notifier.php @@ -379,11 +379,27 @@ function notifier_run($argv, $argc){ dbesc($recip_str) ); - // delivery loop require_once('include/salmon.php'); + $interval = intval(get_config('system','delivery_interval')); + if(! $interval) + $interval = 2; + + // delivery loop + if(count($r)) { + + foreach($r as $contact) { + if((! $mail) && (! $fsuggest) && (! $followup) && (! $contact['self'])) { + q("insert into deliverq ( `cmd`,`item`,`contact` ) values ('%s', %d, %d )", + dbesc($cmd), + intval($item_id), + intval($contact['id']) + ); + } + } + foreach($r as $contact) { if($contact['self']) continue; @@ -392,13 +408,8 @@ function notifier_run($argv, $argc){ // we will deliver single recipient types of message and email receipients here. if((! $mail) && (! $fsuggest) && (! $followup)) { - $interval = intval(get_config('system','delivery_interval')); - if(! $interval) - $interval = 2; - proc_run('php','include/delivery.php',$cmd,$item_id,$contact['id']); - sleep($interval); - continue; + @time_sleep_until(microtime(true) + (float) $interval); } $deliver_status = 0; @@ -624,6 +635,18 @@ function notifier_run($argv, $argc){ if(count($r)) { logger('pubdeliver: ' . print_r($r,true)); + // throw everything into the queue in case we get killed + + foreach($r as $rr) { + if((! $mail) && (! $fsuggest) && (! $followup)) { + q("insert into deliverq ( `cmd`,`item`,`contact` ) values ('%s', %d, %d )", + dbesc($cmd), + intval($item_id), + intval($rr['id']) + ); + } + } + foreach($r as $rr) { /* Don't deliver to folks who have already been delivered to */ @@ -634,13 +657,9 @@ function notifier_run($argv, $argc){ } if((! $mail) && (! $fsuggest) && (! $followup)) { - $interval = intval(get_config('system','delivery_interval')); - if(! $interval) - $interval = 2; - + logger('notifier: delivery agent: ' . $rr['name'] . ' ' . $rr['id']); proc_run('php','include/delivery.php',$cmd,$item_id,$rr['id']); - sleep($interval); - continue; + @time_sleep_until(microtime(true) + (float) $interval); } } } diff --git a/include/queue.php b/include/queue.php index f1bcf2e9..0cb6fcec 100644 --- a/include/queue.php +++ b/include/queue.php @@ -38,6 +38,20 @@ function queue_run($argv, $argc){ logger('queue: start'); + $interval = intval(get_config('system','delivery_interval')); + if(! $interval) + $interval = 2; + + + $r = q("select * from deliverq where 1"); + if(count($r)) { + foreach($r as $rr) { + logger('queue: deliverq'); + proc_run('php','include/delivery.php',$rr['cmd'],$rr['item'],$rr['contact']); + @time_sleep_until(microtime(true) + (float) $interval); + } + } + $r = q("SELECT `queue`.*, `contact`.`name`, `contact`.`uid` FROM `queue` LEFT JOIN `contact` ON `queue`.`cid` = `contact`.`id` WHERE `queue`.`created` < UTC_TIMESTAMP() - INTERVAL 3 DAY"); diff --git a/update.php b/update.php index c23ce95c..14bc48ab 100644 --- a/update.php +++ b/update.php @@ -1,6 +1,6 @@