}
}
-
+ var $queue = array();
+ var $domain_queue = array(); // false to use nextquee
+ var $next_queue = array();
+ var $server_id;
+
function get($r,$opts=array())
{
$this->parseArgs($opts);
$w = DB_DataObject::factory($this->table);
$total = 0;
+
+
+ $ff = HTML_FlexyFramework::get();
+ if (!empty($ff->Core_Notify['servers'])) {
+ if (!isset($ff->Core_Notify['servers'][gethostname()])) {
+ $this->jerr("Core_Notify['servers']['" . gethostname() ."'] is not set");
+ }
+ $w->server_id = array_search(gethostname(),array_keys($ff->Core_Notify['servers']));
+ }
+ if (!empty($this->evtype)) {
+ $w->evtype = $this->evtype;
+ }
+
+
+
if (!empty($opts['old'])) {
// show old and new...
$w->limit($opts['limit']); // we can run 1000 ...
}
- if (!empty($this->evtype)) {
- $w->evtype = $this->evtype;
- }
- $ff = HTML_FlexyFramework::get();
- if (!empty($ff->Core_Notify['servers'])) {
- if (!isset($ff->Core_Notify['servers'][gethostname()])) {
- $this->jerr("Core_Notify['servers']['" . gethostname() ."'] is not set");
- }
- $w->server_id = array_search(gethostname(),array_keys($ff->Core_Notify['servers']));
- }
-
- $this->addFilter($w);
-
+
$w->autoJoin();
- $w->find();
+ $total = $w->find();
+
- $ar = array(); // $w->fetchAll();
if (!empty($opts['list'])) {
}
//echo "BATCH SIZE: ". count($ar) . "\n";
- $pushed = array();
- $requeue = array();
+
+
while (true) {
// only add if we don't have any queued up..
- if (empty($ar) && $w->fetch()) {
- $ar[] = clone($w);
+ if (empty($this->queue) && $w->fetch()) {
+ $this->queue[] = clone($w);
$total--;
}
+
+ $this->logecho("BATCH SIZE: Queue=". count($this->queue) . " TOTAL = " . $total );
- $this->logecho("BATCH SIZE: ". (count($ar) + $total) );
-
- if (empty($ar)) {
- $this->logecho("COMPLETED MAIN QUEUE - running deleted");
-
- if (empty($pushed)) {
- break;
+ if (empty($this->queue)) {
+ $this->logecho("COMPLETED MAIN QUEUE - running maxed out domains");
+ if ($this->domain_queue !== false) {
+ $this->queue = $this->remainingDomainQueue();
+
+ continue;
}
- $ar = $pushed;
- $pushed = false;
- continue;
+ break; // nothing more in queue.. and no remaining one
}
- $p = array_shift($ar);
+ $p = array_shift($this->queue);
if (!$this->poolfree()) {
- array_unshift($ar,$p); /// put it back on..
+ array_unshift($this->queue,$p); /// put it back on..
sleep(3);
continue;
}
- $email = $p->person() ? $p->person()->email : $p->to_email;
+ // not sure what happesn if person email and to_email is empty!!?
+ $email = empty($p->to_email) ? ($p->person() ? $p->person()->email : $p->to_email) : $p->to_email;
+
+ $black = $this->isBlacklisted($email);
+ if ($black !== false) {
+ $this->logecho("DOMAIN blacklisted - {$email} - moving to another pool");
+ $this->updateServer($p, $black);
+ continue;
+ }
+
if ($this->poolHasDomain($email) > $this->max_to_domain) {
- if ($pushed === false) {
- // we only try once to requeue..
- $requeue[] = $p;
- continue;
- }
- $pushed[] = $p;
-
+ // push it to a 'domain specific queue'
+ $this->logecho("REQUEING - maxed out that domain - {$email}");
+ $this->pushQueueDomain($p, $email);
+
//sleep(3);
continue;
}
+ $this->logecho("REQUEUING all emails that maxed out:" . count($this->next_queue));
+ if (!empty($this->next_queue)) {
+
+ foreach($this->next_queue as $p) {
+ $this->updateServer($p);
+ }
+ }
+
+ $this->logecho("QUEUE COMPLETE - waiting for pool to end");
// we should have a time limit here...
while(count($this->pool)) {
$this->poolfree();
- sleep(3);
+ sleep(3);
}
- foreach($requeue as $p) {
- $pp = clone($p);
- $p->act_when = $p->sqlValue('NOW + INTERVAL 1 MINUTE');
- $p->update($pp);
-
- }
+
$this->logecho("DONE");
}
+ function isBlacklisted($email)
+ {
+ // return current server id..
+ $ff = HTML_FlexyFramework::get();
+ //$this->logecho("CHECK BLACKLISTED - {$email}");
+ if (empty($ff->Core_Notify['servers'])) {
+ return false;
+ }
+
+ if (!isset($ff->Core_Notify['servers'][gethostname()]['blacklisted'])) {
+ return false;
+ }
+
+ // get the domain..
+ $ea = explode('@',$email);
+ $dom = strtolower(array_pop($ea));
+
+ //$this->logecho("CHECK BLACKLISTED DOM - {$dom}");
+ if (!in_array($dom, $ff->Core_Notify['servers'][gethostname()]['blacklisted'] )) {
+ return false;
+ }
+ //$this->logecho("RETURN BLACKLISTED TRUE");
+ return array_search(gethostname(),array_keys($ff->Core_Notify['servers']));
+ }
+
+ // this sequentially distributes requeued emails.. - to other servers. (can exclude current one if we have that flagged.)
+ function updateServer($w, $exclude = -1)
+ {
+ $ff = HTML_FlexyFramework::get();
+ static $num = 0;
+ if (empty($ff->Core_Notify['servers'])) {
+ return;
+ }
+ $num = ($num+1) % count(array_keys($ff->Core_Notify['servers']));
+ if ($exclude == $num ) {
+ $num = ($num+1) % count(array_keys($ff->Core_Notify['servers']));
+ }
+ // next server..
+ $pp = clone($w);
+ $w->server_id = $num;
+
+ $w->act_when = $w->sqlValue('NOW + INTERVAL 1 MINUTE');
+ $w->update($pp);
+
+
+ }
function generateNotifications()
return;
}
+ if (!isset($ff->Core_Notify['servers'][gethostname()])) {
+ $this->jerr("Core_Notify['servers']['" . gethostname() ."'] is not set");
+ }
+ // only run this on the main server...
+ if (array_search(gethostname(),array_keys($ff->Core_Notify['servers'])) > 0) {
+ return;
+ }
+
$num_servers = count(array_keys($ff->Core_Notify['servers']));
$p = DB_DataObject::factory($this->table);
+ $p->whereAdd("
+ sent < '2000-01-01'
+ and
+ event_id = 0
+ and
+ act_start < NOW() + INTERVAL 3 HOUR
+ and
+ server_id < 0"
+
+ );
+ if ($p->count() < 1) {
+ return;
+ }
+ $p = DB_DataObject::factory($this->table);
// 6 seconds on this machne...
$p->query("
UPDATE
and
event_id = 0
and
- act_start < NOW()
+ act_start < NOW() + INTERVAL 3 HOUR
and
server_id < 0
ORDER BY
id ASC
LIMIT
- 20000
+ 10000
");
$pipe = array();
- $this->logecho("call proc_open $cmd");
+ //$this->logecho("call proc_open $cmd");
if ($this->max_pool_size === 1) {
+ $this->logecho("call passthru [{$email}] $cmd");
passthru($cmd);
return;
}
);
- $this->logecho("RUN ({$info['pid']}) $cmd ");
+ $this->logecho("RUN [{$email}] ({$info['pid']}) $cmd ");
}
function poolfree()
//fclose($p['pipes'][1]);
fclose($p['pipes'][0]);
fclose($p['pipes'][2]);
- $this->logecho("TERMINATING: ({$p['pid']}) " . $p['cmd'] . " : " . file_get_contents($p['out']));
+ $this->logecho("TERMINATING: ({$p['pid']}) {$p['email']} " . $p['cmd'] . " : " . file_get_contents($p['out']));
@unlink($p['out']);
// schedule again
// $pool[] = $p;
// continue;
//}
- $this->logecho("ENDED: ({$p['pid']}) " . $p['cmd'] . " : " . file_get_contents($p['out']) );
+ $this->logecho("ENDED: ({$p['pid']}) {$p['email']} " . $p['cmd'] . " : " . file_get_contents($p['out']) );
@unlink($p['out']);
+ // at this point we could pop onto the queue the
+ $this->popQueueDomain($p['email']);
+
//unlink($p['out']);
}
$this->logecho("POOL SIZE: ". count($pool) );
return $ret;
}
+ function popQueueDomain($email)
+ {
+ $ea = explode('@',$email);
+ $dom = strtolower(array_pop($ea));
+ if (empty($this->domain_queue[$dom])) {
+ return;
+ }
+ array_unshift($this->queue, array_shift($this->domain_queue[$dom]));
+
+ }
+
+ function pushQueueDomain($e, $email)
+ {
+ if ($this->domain_queue === false) {
+ $this->next_queue[] = $e;
+ return;
+ }
+
+ $ea = explode('@',$email);
+ $dom = strtolower(array_pop($ea));
+ if (!isset($this->domain_queue[$dom])) {
+ $this->domain_queue[$dom] = array();
+ }
+ $this->domain_queue[$dom][] = $e;
+ }
+ function remainingDomainQueue()
+ {
+ $ret = array();
+ foreach($this->domain_queue as $dom => $ar) {
+ $ret = array_merge($ret, $ar);
+ }
+ $this->domain_queue = false;
+ return $ret;
+ }
+
+
function output()
{
projects / Pman.Core / blobdiff