Ego_Queue.php

gehe zur Dokumentation dieser Datei

<?php
class Ego_Queue {
    private static $queue = array(); 
    private static $queueMd5 = array(); 
    private static $max = 0; 
    private static $start = 0; 
    private static $execStart = 0;
    private static $stats = [];
    private static $executing = false;
 
    public static function size() {
        return sizeof(Ego_Queue::$queue);
    }
 
    public static function setExecuting($b) {
        self::$executing = $b;
    }
 
    private static function getHash($entry) {
        return md5(print_r($entry, true));
    }
 
    public static function add(callable $call, $params = array(), $first = false) {
        $entry = array(
            'call' => $call,
            'params' => $params
        );
 
        // Falls die Queue bereits abgearbeitet wird, werden alle neuen Einträge in der Queue sofort ausgeführt
        if (self::$executing) {
            self::call($entry);
            return;
        }
 
        // Eine mehrfache Ausführung der gleichen Methode vermeiden.
        $md5 = self::getHash($entry);
        if (self::$queueMd5[$md5]) {
            return;
        }
        self::$queueMd5[$md5] = true;
 
        if ($first) {
            array_unshift(self::$queue, $entry);
        } else {
            array_push(self::$queue, $entry);
        }
 
        self::log('add', $entry);
    }
 
    public static function remove(callable $call, $params = array()) {
        if (sizeof(self::$queue) > 0) {
            $entry = array(
                'call' => $call,
                'params' => $params
            );
            foreach (self::$queue as $offset => $queue) {
                if ($queue === $entry) {
                    array_splice(self::$queue, $offset, 1);
                    unset(self::$queueMd5[self::getHash($entry)]);
                    break;
                }
            }
        }
    }
 
    public static function exec($close_request = true, $flush=true) {
        if (sizeof(self::$queue) <= 0) { // Keine Queue vorhanden.
            return;
        }
        self::$executing = true;
 
        if ($flush) {
            Ego_System::flush(); // Zunächst alles ausgeben
        }
        if ( $close_request && function_exists('fastcgi_finish_request') ) {
            fastcgi_finish_request(); // und dann die Verbindung schließen und trotzdem weitermachen.
        }
        self::$start = microtime(true);
 
        if ($GLOBALS['egotec_conf']['monitor']) { // Monitor Queue Aufrufe
            apcu_inc('queue_current_exec'); // Anzahl aktuell laufender Queues.
            apcu_inc('queue_count_exec'); // Anzahl Queues gesamt.
            apcu_inc('queue_size_exec', sizeof(self::$queue)); // Anzahl Queues gesamt.
        }
 
        while ($entry = array_shift(self::$queue)) {
            self::call($entry);
        }
 
        if ($GLOBALS['egotec_conf']['monitor']) { // Monitor Queue Aufrufe
            $end = microtime(true);
            $d = $end - self::$start;
            apcu_dec('queue_current_exec'); // Anzahl aktuell laufender Queues.
            apcu_inc('queue_duration_exec', $d); // Anzahl Queues gesamt.
        }
    }
 
    private static function call($entry) {
        if (@is_callable($entry['call'])) {
            self::log('execStart', $entry);
            try {
                call_user_func_array($entry['call'], $entry['params']);
            } catch (Exception $e) {
                egotec_error_log(
                    "queue call exception\n".
                    $e->getMessage()."\n".
                    $e->getFile()." ".$e->getLine()."\n".
                    $e->getTraceAsString()
                );
            }
            self::log('execEnd', $entry);
        }
    }
 
    private static function log($method, $entry) {
        if (!$GLOBALS['egotec_conf']['log_queue']) return;
 
        if ($method=='execStart') {
            self::$execStart = microtime(true);
        }
 
        $s = "Ego_Queue::$method\n";
        if (is_array($entry['call'])) {
            $name2 = ' ';
            if (is_string($entry['call'][0])) {
                $name = $entry['call'][0].' '.$entry['call'][1];
            } else {
                $class = get_class($entry['call'][0]);
                switch ($class) {
                    case 'Site':
                        $name2.= $entry['call'][0]->name;
                        break;
                    case 'Ego_Cache_apcu':
                    case 'Ego_Cache_custom':
                    case 'Ego_Cache_file':
                        $name2.= $entry['call'][0]->getPath();
                        break;
                    case 'Ego_Search_Lucene':
                        $name2.= $entry['call'][0]->getConfig()['table'];
                        break;
                    case 'Ego_Search_Elastic':
                        $name2.= $entry['call'][0]->getConfig()['index'];
                        break;
                    default:
                }
                $name = $class." ".print_r($entry['call'][1], true);
            }
        } elseif (is_object($entry['call'])) {
            $name = get_class($entry['call']);
        } else {
            $name = $entry['call'];
        }
        $s.= $name.$name2."\n";
 
        if ($method=='add') {
            self::$stats[$name]['count']++;
        }
 
        if ($method=='execEnd') {
            self::$stats[$name]['time']+= microtime(true)-self::$execStart;
            uasort(self::$stats, function($a, $b) { return $b['time']-$a['time']; } );
        }
 
        $size = sizeof(self::$queue);
        if ($size>self::$max) {
            self::$max = $size;
        }
        $d = self::$start ? microtime(true) - self::$start:0;
        file_put_contents(
            $GLOBALS['egotec_conf']['log_dir'] . 'q' . getmypid(),
            $size . ' ' . self::$max . ' ' . round($d, 3) . ' ' . $s . "\n" .
            print_r(self::$stats, true)
        );
    }
}
register_shutdown_function(array('Ego_Queue', 'exec'));