--- a/busui/owa/modules/base/classes/fileEventQueue.php +++ b/busui/owa/modules/base/classes/fileEventQueue.php @@ -1,1 +1,274 @@ - +<?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 © 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; + } +} + +?>