<?php
 
//
// Open Web Analytics - An Open Source Web Analytics Framework
//
// Copyright 2006 Peter Adams. All rights reserved.
//
// Licensed under GPL v2.0 http://www.gnu.org/copyleft/gpl.html
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// $Id$
//
 
require_once(OWA_BASE_CLASS_DIR.'eventQueue.php');
require_once(OWA_BASE_CLASS_DIR.'event.php');
require_once(OWA_PEARLOG_DIR . DIRECTORY_SEPARATOR . 'Log.php');
require_once(OWA_PEARLOG_DIR . DIRECTORY_SEPARATOR . 'Log/file.php');
 
/**
 * File based Event Queue Implementation
 * 
 * @author      Peter Adams <peter@openwebanalytics.com>
 * @copyright   Copyright &copy; 2006 Peter Adams <peter@openwebanalytics.com>
 * @license     http://www.gnu.org/copyleft/gpl.html GPL v2.0
 * @category    owa
 * @package     owa
 * @version             $Revision$            
 * @since               owa 1.0.0
 */
 
class owa_fileEventQueue extends owa_eventQueue {
        
        var $queue;
        var $error_logger;
        var $queue_dir;
        var $event_file;
        
        function __construct($queue_dir = '') {
                
                // set event file
                if (!$queue_dir) {
                        $this->queue_dir = owa_coreAPI::getSetting('base', 'async_log_dir');
                }
                
                $this->event_file = $this->queue_dir.'events.txt';
                $this->lock_file = $this->queue_dir.'lock.txt';
        }
                
        function makeQueue() {
                
                //make file queue
                $conf = array('mode' => 0600, 'timeFormat' => '%X %x');
                //$this->queue = &Log::singleton('async_queue', $this->event_file, 'async_event_queue', $conf);
                $this->queue = Log::singleton('file', $this->event_file, 'async_event_queue', $conf);
                $this->queue->_lineFormat = '%1$s|*|%2$s|*|[%3$s]|*|%4$s';
                // not sure why this is needed but it is.
                $this->queue->_filename = $this->event_file;
        }
        
        function addToQueue($event) {
                
                if (!$this->queue) {
                        $this->makeQueue();
                }
                
                $this->queue->log(urlencode(serialize($event)));
        
        }
        
        function processQueue($event_file = '') {
        
                if ($event_file) {
                
                        $this->event_file = $this->queue_dir.$event_file;
                }
                
                if ( file_exists( $this->event_file ) ) {
                        
                        $event_log_rotate_size = owa_coreAPI::getSetting( 'base', 'async_log_rotate_size' );
                        
                        if ( filesize( $this->event_file ) > $event_log_rotate_size ) {
                                
                                owa_coreAPI::notice(sprintf('Starting Async Event Processing Run for: %s', $this->event_file));
                                
                                //check for lock file
                                if (!$this->isLocked()) {
                                        
                                        return $this->process_event_log($this->event_file);
                                        
                                } else {
                                        
                                        owa_coreAPI::notice(sprintf('Previous Process (%d) still active. Terminating Run.', $former_pid));
                                }
                                                        
                        } else {
                                
                                owa_coreAPI::debug("Event file is not large enough to process yet. Size is only: ".filesize($this->event_file));
                        }
                        
                } else {
                        
                        owa_coreAPI::debug("No event file found at: ".$this->event_file);
                }
                                
        }
        
        function isLocked() {
                
                if (file_exists($this->lock_file)) {
                        //read contents of lock file for last PID
                        $lock = fopen($this->lock_file, "r") or die ("Could not read lock file");
                        if ($lock) {
                                while (!feof($lock)) {
                                        $former_pid = fgets($lock, 4096);
                                }
                                fclose($lock);
                        }
                        
                        //check to see if former process is still running
                        $ps_check = $this->isRunning($former_pid);
                        //if the process is still running, exit.
                        if ($ps_check) {
                                owa_coreAPI::notice(sprintf('Previous Process (%d) still active. Terminating Run.', $former_pid));
                                return true;
                        //if it's not running remove the lock file and proceead.
                        } else {
                                owa_coreAPI::debug(sprintf('Process %d is no longer running. Deleting old Lock file. \n', $former_pid));
                                unlink ($this->lock_file);
                                return false;
                        }
        
                } else {
                        return false;   
                }
        }
        
        function isRunning($pid) {
                
                $process_state = '';
      
                exec("ps $pid", $process_state);
                //print $pid;
                print_r($process_state);
   
                if (count($process_state) >= 2) {
                        return true;
                } else {
                        return false;
                }
        }
        
        function process_event_log($file) {
                
                // check to see if event log file exisits
                if (!file_exists($file)) {
                        owa_coreAPI::debug("Event file does not exist at $file");
                        return false;
                }
                        
                //create lock file
                $this->create_lock_file();
                
                // get event dispatcher
                $dispatch = owa_coreAPI::getEventDispatch();
                
                // Create a new log file name   
                $new_file_name = $this->queue_dir.time().".".getmypid();
                $new_file = $new_file_name.".processing";
                
                // Rename current log file 
                rename ($file, $new_file ) or die ("Could not rename file");
                owa_coreAPI::debug('renamed event file.');
                
                // open file for reading
                $handle = @fopen($new_file, "r");
                if ($handle) {
                        while (!feof($handle)) {
                                
                                // Read row
                                $buffer = fgets($handle, 14096); // big enough?
                                        
                                // Parse the row
                                $event = $this->parse_log_row($buffer);
                                
                                // Log event to the event queue
                                if (!empty($event)) {
                                        //print_r($event);
                                        // debug
                                        owa_coreAPI::debug(sprintf('Processing: %s (%s)', '', $event->guid));
                                        // send event object to event queue
                                        $ret = $dispatch->notify($event);
                                        
                                        // is the dispatch was not successful then add the event back into the queue.
                                        if ( $ret != OWA_EHS_EVENT_HANDLED ) {
                                                $dispatch->asyncNotify($event);
                                        }
                                        
                                } else {
                                        owa_coreAPI::debug("No event found in log row. Must be end of file.");
                                }                                               
                        }
                        //Close file
                        fclose($handle);
                        
                        // rename file to mark it as processed
                        $processed_file_name = $new_file_name.".processed";
                        rename ($new_file, $processed_file_name) or die ("Could not rename file");      
                        owa_coreAPI::debug(sprintf('Processing Complete. Renaming File to %s', $processed_file_name ));
                        
                        //Delete processed file
                        unlink($processed_file_name);
                        owa_coreAPI::debug(sprintf('Deleting File %s', $processed_file_name));
                        
                        //Delete Lock file
                        unlink($this->lock_file);
                        
                        return true;    
                } else {
                        //could not open file for processing
                        owa_coreAPI::error(sprintf('Could not open file %s. Terminating Run.', $new_file));
                }
        }
 
        function makeErrorLogFile() {
                
                $conf = array('mode' => 640, 'timeFormat' => '%X %x');
                $this->error_logger = &Log::singleton('file', owa_coreAPI::getSetting('async_error_log_file'), 'ident', $conf);
                $this->error_logger->_lineFormat = '[%3$s]';
                $this->error_logger->_filename = owa_coreAPI::getSetting('async_error_log_file');
        }
        
        function logError($event) {
        
        }
        
        /**
         * Parse row from event log file
         *
         * @param string $row
         * @return array
         */
        function parse_log_row($row) {
                if ($row) {
                        $raw_event = explode("|*|", $row);
                        //print_r($raw_event);
                        //$row_array = array( 'timestamp'               => $raw_event[0], 'event_type'  => $raw_event[3], 'event_obj'           => $raw_event[4]); 
                        $row_array = array( 'timestamp' => $raw_event[0], 'event_obj' => $raw_event[3]); 
                        //print_r($row_array);                  
                        $event = unserialize(urldecode($row_array['event_obj']));
                        //print_r($event);
                        return $event;
                }
        }
        
        function create_lock_file() {
                
                $lock_file = fopen($this->lock_file, "w+") or die ("Could not create lock file at: ".$this->lock_file);
                                                                
                // Write PID to lock file
                if (fwrite($lock_file, getmypid()) === FALSE) {
                owa_coreAPI::debug('Cannot write to lock file. Terminating Run.');
                exit;
                }
                
                return;
        }
}
 
?>