3 * Table Definition for core_notify_server
5 class_exists('DB_DataObject') ? '' : require_once 'DB/DataObject.php';
7 class Pman_Core_DataObjects_Core_notify_server extends DB_DataObject
10 /* the code below is auto generated do not remove the above tag */
12 public $__table = 'core_notify_server'; // table name
13 public $id; // int(11) not_null primary_key auto_increment
22 function applyFilters($q, $au, $roo)
24 if (isset($q['_with_queue_size'])) {
25 $this->addQueueSize();
30 function addQueueSize()
32 // look for database links for server_id (which should find core_notify + others..)
33 $cn = get_class(DB_DataObject::factory('core_notify'));
35 foreach($this->databaseLinks() as $tbl => $kv) {
36 foreach($kv as $k=>$v) {
37 if ($v != 'core_notify_server:id') {
41 $test = DB_DAtaObject::factory($tbl);
42 if (!is_a($test, $cn)) {
49 die("OOPS - no tables for notify_server references");
52 foreach($tables as $t) {
59 server_id = core_notify_server.id
61 sent < '1970-01-01' OR sent IS NULL
67 $this->selectAdd("(" . implode(" + ", $totals) . ") as in_queue ");
76 // most services should call this first..
78 function getCurrent($notify, $force = false)
80 static $current = false;;
82 if ($current !== false) {
86 $ns = DB_DataObject::factory('core_notify_server');
88 $ns->poolname = $notify->poolname;
90 $ns->hostname = gethostname();
92 if ($ns->find(true)) {
97 $notify->jerr("Server not found for this server " . gethostname() . " in core_notify_server" );
99 // fallback to any server - if we are using force. (this is so helo will work...)
101 $ns = DB_DataObject::factory('core_notify_server');
103 $ns->hostname = gethostname();
104 if (!$ns->find(true)) {
105 $notify->jerr("Server not found for this server " . gethostname() . " in core_notify_server" );
112 function isFirstServer()
114 $servers = $this->availableServers();
115 if (empty($servers)) {
118 // only run this on the first server...
119 return $this->id == $servers[0]->id;
123 // called on current server.
124 function assignQueues($notify)
128 $servers = $this->availableServers();
131 foreach($servers as $s) {
137 $notify->jerr("no configured servers in core_notify_server for poolname = {$notify->poolname}");
141 // only run this on the first server...
142 if ($this->id != $ids[0]) {
145 foreach($ids as $rn) {
149 $num_servers = count($ids);
151 if ($num_servers == 1) {
152 $p = DB_DataObject::factory($notify->table);
157 server_id = {$ids[0]}
163 act_start < NOW() + INTERVAL 3 HOUR
165 server_id != {$ids[0]}
172 $p = DB_DataObject::factory($notify->table);
178 act_start < NOW() + INTERVAL 3 HOUR
180 server_id NOT IN (" . implode(",", $ids) . ")
182 $p->orderBy('act_when asc'); //?
183 $total_add = $p->count();
184 if ($total_add < 1) {
188 $to_add = $p->fetchAll('id');
190 $p = DB_DataObject::factory($notify->table);
197 server_id IN (" . implode(",", $ids) . ")
200 $p->selectAdd('server_id, count(id) as n');
201 $p->groupBy('server_id');
202 $in_q = $p->fetchAll('server_id', 'n');
204 // if queue is empty it will not get allocated anything.
205 foreach($ids as $sid) {
206 if (!isset($in_q[$sid])) {
211 foreach($in_q as $sid => $n) {
217 $target_len = floor( ($totalq + $total_add) / $num_servers );
219 foreach($in_q as $sid => $cq) {
220 if ( $cq > $target_len) {
223 $up[ $sid ] = array_slice($to_add, 0, $target_len - $cq);
226 // add the reminder evently
227 foreach($to_add as $n=>$i) {
229 $up[ $ids[$n % $num_servers] ][] = $i;
232 // distribution needs to go to ones that have the shortest queues. - so to balance out the queues
236 foreach($up as $sid => $nids) {
240 $p = DB_DataObject::factory($notify->table);
247 id IN (". implode(',', $nids). ')'
251 DB_DataObject::factory("core_notify_blacklist")->prune();
254 // called on current server.
256 function availableServers()
258 $ns = DB_DataObject::factory('core_notify_server');
259 $ns->poolname = $this->poolname;
261 $ns->orderBy('id ASC');
262 return $ns->fetchAll();
266 function updateNotifyToNextServer( $cn , $when = false, $allow_same = false)
268 // fixme - this should take into account blacklisted - and return false if no more servers are available
269 $email = empty($cn->to_email) ? ($cn->person() ? $cn->person()->email : $cn->to_email) : $cn->to_email;
271 $w = DB_DataObject::factory($cn->tableName());
274 $servers = $this->availableServers();
276 foreach($servers as $i => $s) {
277 if ($s->id == $this->id) {
282 $offset = ($start + 1) % count($servers);
284 while ($offset != $start) {
285 $s = $servers[$offset];
286 if (!$s->isBlacklisted($email)) {
290 $offset = ($offset + 1) % count($servers);
293 if ($good == false && $allow_same) {
297 if ($good == false) {
304 $w->server_id = $good->id;
305 $w->act_when = $when === false ? $w->sqlValue('NOW() + INTERVAL 1 MINUTE') : $when;
311 function isBlacklisted($email)
313 // return current server id..
314 static $cache = array();
316 $ea = explode('@',$email);
317 $dom = strtolower(array_pop($ea));
318 if (isset( $cache[$this->id . '-'. $dom])) {
319 return $cache[$this->id . '-'. $dom];
322 $cd = DB_DataObject::factory('core_domain')->loadOrCreate($dom);
324 $bl = DB_DataObject::factory('core_notify_blacklist');
325 $bl->server_id = $this->id;
326 $bl->domain_id = $cd->id;
328 $cache[$this->id . '-'. $dom] = true;
336 $ff = HTML_FlexyFramework::get();
337 $ff->Mail['helo'] = $this->helo;
340 function checkSmtpResponse($errmsg, $core_domain)
342 $bl = DB_DataObject::factory('core_notify_blacklist');
343 $bl->server_id = $this->id;
344 $bl->domain_id = $core_domain->id;
348 // is it a blacklist message
349 if (!$bl->messageIsBlacklisted($errmsg)) {
352 $bl->error_str = $errmsg;
353 $bl->added_dt = $bl->sqlValue("NOW()");