git.roojs.org Git - Pman.Core/blob - DataObjects/Core_notify_server.php

   1 <?php
   2 /**
   3  * Table Definition for core_notify_server
   4  */
   5 class_exists('DB_DataObject') ? '' : require_once 'DB/DataObject.php';
   6
   7 class Pman_Core_DataObjects_Core_notify_server extends DB_DataObject
   8 {
   9     ###START_AUTOCODE
  10     /* the code below is auto generated do not remove the above tag */
  11
  12     public $__table = 'core_notify_server';    // table name
  13     public $id;                              // int(11)  not_null primary_key auto_increment
  14     public $hostname;
  15     public $helo;
  16     public $poolname;
  17     public $is_active;
  18     public $last_send;
  19
  20
  21
  22     function  applyFilters($q, $au, $roo)
  23     {
  24         if (isset($q['_with_queue_size'])) {
  25             $this->addQueueSize();
  26         }
  27     }
  28
  29
  30     function addQueueSize()
  31     {
  32         // look for database links for server_id (which should find core_notify + others..)
  33         $cn = get_class(DB_DataObject::factory('core_notify'));
  34         $tables = array();
  35         foreach($this->databaseLinks() as $tbl => $kv) {
  36             foreach($kv as $k=>$v) {
  37                 if ($v != 'core_notify_server:id') {
  38                     continue;
  39                 }
  40
  41                 $test = DB_DAtaObject::factory($tbl);
  42                 if (!is_a($test, $cn)) {
  43                     break;
  44                 }
  45                 $tables[] = $tbl;
  46             }
  47         }
  48         if (empty($tables)) {
  49             die("OOPS - no tables for notify_server references");
  50         }
  51         $totals = array();
  52         foreach($tables as $t) {
  53             $totals[] = "
  54                 COALESCE((SELECT
  55                     count(id)
  56                 FROM
  57                     $t
  58                 WHERE
  59                     server_id = core_notify_server.id
  60                 AND
  61                     sent < '1970-01-01' OR sent IS NULL
  62                  and
  63                         event_id = 0
  64                 ),0)
  65             ";
  66         }
  67         $this->selectAdd("(" . implode(" + ", $totals) . ") as in_queue ");
  68
  69
  70
  71
  72
  73     }
  74
  75
  76     // most services should call this first..
  77
  78     function getCurrent($notify, $force = false)
  79     {
  80         static $current = false;;
  81
  82         if ($current !== false) {
  83             return $current;
  84         }
  85
  86         $ns = DB_DataObject::factory('core_notify_server');
  87         if (!$ns->count()) {
  88             $ns->id = 0;
  89             return $ns;
  90         }
  91
  92
  93         $ns->poolname = $notify->poolname;
  94         $ns->is_active = 1;
  95         $ns->hostname = gethostname();
  96         $ns->limit(1);
  97         if ($ns->find(true)) {
  98             $current = $ns;
  99             return $ns;
 100         }
 101         if (!$force) {
 102             $notify->jerr("Server not found for this server " .  gethostname() . " in core_notify_server" );
 103         }
 104         // fallback to any server - if we are using force. (this is so helo will work...)
 105
 106         $ns = DB_DataObject::factory('core_notify_server');
 107         $ns->is_active = 1;
 108         $ns->hostname = gethostname();
 109         if (!$ns->find(true)) {
 110             $notify->jerr("Server not found for this server " .  gethostname() . " in core_notify_server" );
 111         }
 112         $current = $ns;
 113         return $ns;
 114     }
 115
 116
 117     function isFirstServer()
 118     {
 119         if (!$this->id) {
 120             return true;
 121         }
 122
 123         $servers = $this->availableServers();
 124         if (empty($servers)) {
 125             return false;
 126         }
 127         // only run this on the first server...
 128         return $this->id == $servers[0]->id;
 129     }
 130
 131
 132     // called on current server.
 133     function assignQueues($notify)
 134     {
 135         if (!$this->id) {
 136             return true;
 137         }
 138
 139         $servers = $this->availableServers();
 140         $ids = array();
 141         $up = array();
 142         foreach($servers as $s) {
 143             $ids[] = $s->id;
 144         }
 145
 146
 147         if (empty($ids)) {
 148             $notify->jerr("no configured servers in core_notify_server for poolname = {$notify->poolname}");
 149
 150         }
 151
 152         // only run this on the first server...
 153         if ($this->id != $ids[0]) {
 154             return;
 155         }
 156         foreach($ids as $rn) {
 157             $up[$rn]  = array();
 158         }
 159
 160         $num_servers = count($ids);
 161
 162         if ($num_servers == 1) {
 163             $p = DB_DataObject::factory($notify->table);
 164             $p->query("
 165                 UPDATE
 166                     {$notify->table}
 167                 SET
 168                     server_id = {$ids[0]}
 169                 WHERE
 170                     sent < '2000-01-01'
 171                     and
 172                     event_id = 0
 173                     and
 174                     act_start < NOW() +  INTERVAL 3 HOUR
 175                     and
 176                     server_id != {$ids[0]}
 177             ");
 178             return;
 179         }
 180
 181
 182
 183         $p = DB_DataObject::factory($notify->table);
 184         $p->whereAdd("
 185                 sent < '2000-01-01'
 186                 and
 187                 event_id = 0
 188                 and
 189                 act_start < NOW() +  INTERVAL 3 HOUR
 190                 and
 191                 server_id NOT IN (" . implode(",", $ids) . ")
 192         ");
 193         $p->orderBy('act_when asc'); //?
 194         $total_add = $p->count();
 195         if ($total_add < 1) {
 196             return;
 197         }
 198
 199         $to_add = $p->fetchAll('id');
 200
 201         $p = DB_DataObject::factory($notify->table);
 202         $p->whereAdd("
 203                 sent < '2000-01-01'
 204                 and
 205                 event_id = 0
 206
 207                 and
 208                 server_id IN (" . implode(",", $ids) . ")
 209         ");
 210         $p->selectAdd();
 211         $p->selectAdd('server_id, count(id) as  n');
 212         $p->groupBy('server_id');
 213         $in_q = $p->fetchAll('server_id', 'n');
 214
 215         // if queue is empty it will not get allocated anything.
 216         foreach($ids as $sid) {
 217             if (!isset($in_q[$sid])) {
 218                 $in_q[$sid] = 0;
 219             }
 220         }
 221         $totalq = 0;
 222         foreach($in_q as $sid => $n) {
 223             $totalq += $n;
 224         }
 225
 226
 227         // new average queue
 228         $target_len = floor(  ($totalq + $total_add) / $num_servers );
 229
 230         foreach($in_q as $sid => $cq) {
 231             if ( $cq > $target_len) {
 232                 continue;
 233             }
 234             $up[ $sid ] = array_slice($to_add, 0, $target_len - $cq);
 235         }
 236
 237         // add the reminder evently
 238         foreach($to_add as $n=>$i) {
 239
 240             $up[  $ids[$n % $num_servers] ][] = $i;
 241         }
 242
 243         // distribution needs to go to ones that have the shortest queues. - so to balance out the queues
 244
 245
 246
 247         foreach($up as $sid => $nids) {
 248             if (empty($nids)) {
 249                 continue;
 250             }
 251             $p = DB_DataObject::factory($notify->table);
 252             $p->query("
 253                 UPDATE
 254                     {$notify->table}
 255                 SET
 256                     server_id = $sid
 257                 WHERE
 258                     id IN (". implode(',', $nids). ')'
 259             );
 260         }
 261
 262         DB_DataObject::factory("core_notify_blacklist")->prune();
 263
 264     }
 265         // called on current server.
 266
 267     function availableServers()
 268     {
 269         $ns = DB_DataObject::factory('core_notify_server');
 270         $ns->poolname = $this->poolname;
 271         $ns->is_active = 1;
 272         $ns->orderBy('id ASC');
 273         return  $ns->fetchAll();
 274
 275     }
 276
 277     function updateNotifyToNextServer( $cn , $when = false, $allow_same = false)
 278     {
 279         if (!$this->id) {
 280             return;
 281         }
 282
 283         // fixme - this should take into account blacklisted - and return false if no more servers are available
 284         $email = empty($cn->to_email) ? ($cn->person() ? $cn->person()->email : $cn->to_email) : $cn->to_email;
 285
 286         $w = DB_DataObject::factory($cn->tableName());
 287         $w->get($cn->id);
 288
 289         $servers = $this->availableServers();
 290         $start = 0;
 291         foreach($servers as $i => $s) {
 292             if ($s->id == $this->id) {
 293                 $start = $i;
 294             }
 295         }
 296
 297         $offset = ($start + 1)  % count($servers);
 298         $good = false;
 299         while ($offset  != $start) {
 300             $s = $servers[$offset];
 301             if (!$s->isBlacklisted($email)) {
 302                 $good = $s;
 303                 break;
 304             }
 305             $offset = ($offset + 1)  % count($servers);
 306             var_dump($offset);
 307         }
 308         if ($good == false && $allow_same) {
 309             $good = $this;
 310         }
 311
 312         if ($good == false) {
 313             return false;
 314         }
 315
 316
 317         // next server..
 318         $pp = clone($w);
 319         $w->server_id = $good->id;
 320         $w->act_when = $when === false ? $w->sqlValue('NOW() + INTERVAL 1 MINUTE') : $when;
 321         $w->update($pp);
 322         return true;
 323     }
 324
 325
 326     function isBlacklisted($email)
 327     {
 328         if (!$this->id) {
 329             return false;
 330         }
 331
 332         // return current server id..
 333         static $cache = array();
 334          // get the domain..
 335         $ea = explode('@',$email);
 336         $dom = strtolower(array_pop($ea));
 337         if (isset( $cache[$this->id . '-'. $dom])) {
 338             return  $cache[$this->id . '-'. $dom];
 339         }
 340
 341         $cd = DB_DataObject::factory('core_domain')->loadOrCreate($dom);
 342
 343         $bl = DB_DataObject::factory('core_notify_blacklist');
 344         $bl->server_id = $this->id;
 345         $bl->domain_id = $cd->id;
 346         if ($bl->count()) {
 347             $cache[$this->id . '-'. $dom] = true;
 348             return true;
 349         }
 350
 351         return false;
 352     }
 353     function initHelo()
 354     {
 355         if (!$this->id) {
 356             return;
 357         }
 358         $ff = HTML_FlexyFramework::get();
 359         $ff->Mail['helo'] = $this->helo;
 360
 361     }
 362     function checkSmtpResponse($errmsg, $core_domain)
 363     {
 364         if (!$this->id) {
 365             return false;
 366         }
 367         $bl = DB_DataObject::factory('core_notify_blacklist');
 368         $bl->server_id = $this->id;
 369         $bl->domain_id = $core_domain->id;
 370         if ($bl->count()) {
 371             return true;
 372         }
 373         // is it a blacklist message
 374         if (!$bl->messageIsBlacklisted($errmsg)) {
 375             return false;
 376         }
 377         $bl->error_str = $errmsg;
 378         $bl->added_dt = $bl->sqlValue("NOW()");
 379         $bl->insert();
 380         return true;
 381
 382     }
 383
 384 }