implement delivery queue in case notifier gets killed

diff --git a/boot.php b/boot.php
index 060bd7117f8c75dffd3dafe48f39e15bc42d7e55..595dc1fef8caeeccea367c0cfbfb78b38cedf0da 100644 (file)
--- 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',                    "<br />\r\n"     );
 define ( 'ATOM_TIME',              'Y-m-d\TH:i:s\Z' );

diff --git a/database.sql b/database.sql
index f6ae4c7c609aa75ffd2a1619c8443b63e40c6682..9819914f7842f3496e030ed10b0eeaa9c2bec8f8 100644 (file)
--- 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 0df8ea7e47b08f698e820f71420a4f604ec2b7e9..1f5883c266afbf4bc3cf05790b16bebdb73ffa16 100644 (file)
--- 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 6cba0ecec10d145af9c884c7109188c4a7722ba4..99bc21c0bc4932effa24e4e6bf58bff177cd6ebe 100644 (file)
--- 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) {
         *  </decrypted_header>
         */
 
-       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 e9594cff2c35432ee506351fc49f8b02536b5703..1603dec601c0c10cd6521a40e0405b6034dbd35d 100644 (file)
--- 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 b87aa95b1559c2f1e20d9ad31ea78e2efd95b50c..cde156cbdce3e75659cd24cc0263a43cfb503ddd 100644 (file)
--- 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 f1bcf2e9ffeb42c0277692e372425868e7d09c01..0cb6fcec219f17b7f5e9d552e930d586df767357 100644 (file)
--- 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 c23ce95c6f912f8f88ee076948e174693dacc63d..14bc48ab71871519963af6e02a2ef3d926219fa1 100644 (file)
--- a/update.php
+++ b/update.php
@@ -1,6 +1,6 @@
 <?php
 
-define( 'UPDATE_VERSION' , 1083 );
+define( 'UPDATE_VERSION' , 1084 );
 
 /**
  *
@@ -697,3 +697,13 @@ function update_1082() {
                }
        }
 }
+
+function update_1083() {
+       q("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 ;");
+
+}
\ No newline at end of file