X-Git-Url: http://git.roojs.org/?p=Pman.Core;a=blobdiff_plain;f=Notify.php;h=fd582092af5a14fa15dae313ff2f70dba6e54320;hp=705cb6bcfde3092a46e9190c4e82bb3c9041039a;hb=HEAD;hpb=c7b4d24e21e30a6f9a23c42de500991b0df59f75

diff --git a/Notify.php b/Notify.php
index 705cb6bc..3fd4c5b4 100644
--- a/Notify.php
+++ b/Notify.php
@@ -178,6 +178,12 @@ class Pman_Core_Notify extends Pman
    
     function get($r,$opts=array())    
     {
+        
+       // if ($this->database_is_locked()) {
+        //    die("LATER - DATABASE IS LOCKED");
+        //}
+        
+        
         $this->parseArgs($opts); 
          
         //date_default_timezone_set('UTC');
@@ -206,7 +212,8 @@ class Pman_Core_Notify extends Pman
             $w->evtype = $this->evtype;
         }
         
-        
+        $w->server_id = $this->server->id;
+
         
         if (!empty($opts['old'])) {
             // show old and new...
@@ -224,6 +231,7 @@ class Pman_Core_Notify extends Pman
             if (!$this->force) {
                 $w->whereAdd('act_when < NOW()'); // eg.. not if future..
             }
+            
     
             $w->orderBy('act_when ASC'); // oldest first.
             $total = min($w->count(), $opts['limit']);
@@ -233,7 +241,6 @@ class Pman_Core_Notify extends Pman
         }
         
         
-        
     
         
          
@@ -293,9 +300,14 @@ class Pman_Core_Notify extends Pman
             
             $black = $this->server->isBlacklisted($email);
             if ($black !== false) {
-                
+                $this->logecho("Blacklisted - try giving it to next server");
                 if (false === $this->server->updateNotifyToNextServer($p)) {
-                    $p->updateState("????");
+                    $ev = $this->addEvent('NOTIFY', $p, 'BLACKLISTED  FROM our DB');
+                    // we dont have an althenative server to update it with.
+                    $this->logecho("Blacklisted - next server did not work - try again in 30 mins");
+                    $this->server->updateNotifyToNextServer($w,  date("Y-m-d H:i:s",  strtotime('NOW +  30 MINUTES')),true);
+                   // $this->errorHandler( $ev->remarks);
+                   
                 }
                 
                 continue;
@@ -399,10 +411,12 @@ class Pman_Core_Notify extends Pman
         
         
         $tn =  $this->tempName('stdout', true);
+        $tne =  $this->tempName('stderr', true);
         $descriptorspec = array(
             0 => array("pipe", 'r'),  // stdin is a pipe that the child will read from
             1 => array("file", $tn, 'w'),  // stdout is a pipe that the child will write to
-            2 => array("pipe", "w") // stderr is a file to write to
+            2 => array("file", $tne, 'w'),   // stderr is a file to write to
+          //  2 => array("pipe", "w") // stderr is a file to write to
          );
         
         static $php = false;
@@ -451,6 +465,7 @@ class Pman_Core_Notify extends Pman
                 'proc' => $p,
                 'pid' => $info['pid'],
                 'out' => $tn,
+                'oute' => $tne,
                 'cmd' => $cmd,
                 'email' => $email,
                 'pipes' => $pipes,
@@ -470,6 +485,8 @@ class Pman_Core_Notify extends Pman
         foreach($this->pool as $p) {
              
             //echo "CHECK PID: " . $p['pid'] . "\n";
+            
+            
             $info =  proc_get_status($p['proc']);
             //var_dump($info);
             
@@ -491,9 +508,10 @@ class Pman_Core_Notify extends Pman
                     proc_terminate($p['proc'], 9);
                     //fclose($p['pipes'][1]);
                     fclose($p['pipes'][0]);
-                    fclose($p['pipes'][2]);
-                    $this->logecho("TERMINATING: ({$p['pid']}) {$p['email']} " . $p['cmd'] . " : " . file_get_contents($p['out']));
+                    
+                    $this->logecho("TERMINATING: ({$p['pid']}) {$p['email']} " . $p['cmd'] . " : " . file_get_contents($p['out']) . " : " . file_get_contents($p['oute']));
                     @unlink($p['out']);
+                    @unlink($p['oute']);
                     
                     // schedule again
                     $w = DB_DataObject::factory($this->table);
@@ -512,20 +530,25 @@ class Pman_Core_Notify extends Pman
                 continue;
             }
             fclose($p['pipes'][0]);
-            fclose($p['pipes'][2]);
             //echo "CLOSING: ({$p['pid']}) " . $p['cmd'] . " : " . file_get_contents($p['out']) . "\n";
             //fclose($p['pipes'][1]);
             
             proc_close($p['proc']);
-            
-            
+             sleep(1);
+            clearstatcache();
+            if (file_exists('/proc/'. $p['pid'])) {
+                $this->logecho("proc PID={$p['pid']} still here - trying to wait");
+                pcntl_waitpid($p['pid'], $status, WNOHANG);
+            }
+
             //clearstatcache();
             //if (file_exists('/proc/'.$p['pid'])) {
             //    $pool[] = $p;
             //    continue;
             //}
-            $this->logecho("ENDED: ({$p['pid']}) {$p['email']} " .  $p['cmd'] . " : " . file_get_contents($p['out']) );
+            $this->logecho("ENDED: ({$p['pid']}) {$p['email']} " .  $p['cmd'] . " : " . file_get_contents($p['out']) . " : " . file_get_contents($p['oute']));
             @unlink($p['out']);
+            @unlink($p['oute']);
             // at this point we could pop onto the queue the 
             $this->popQueueDomain($p['email']);
             
@@ -596,6 +619,7 @@ class Pman_Core_Notify extends Pman
     function clearOld()
      {
           if ($this->server->isFirstServer()) {
+            
             $p = DB_DataObject::factory($this->table);
             $p->whereAdd("
                 sent < '2000-01-01'