PHP Classes
elePHPant
Icontem

File: QueueDirParser.php

Recommend this page to a friend!
Stumble It! Stumble It! Bookmark in del.icio.us Bookmark in del.icio.us
  Classes of Daniel Kadosh  >  Multi Threaded Daemon - enhanced  >  QueueDirParser.php  >  Download  
File: QueueDirParser.php
Role: Example script
Content type: text/plain
Description: Sample app - Monitor queue dir and parse files in it
Class: Multi Threaded Daemon - enhanced
Daemon that spawns multiple parallel threads
Author: By
Last change:
Date: 5 years ago
Size: 5,543 bytes
 

Contents

Class file image Download
<?php
/*
 * This deamon monitors a directory for files, and will process each file in
 * its own thread.
 *
 * Sample daemon using Benoit Perroud's MultiThreaded Daemon (MTD)
 * See http://code.google.com/p/phpmultithreadeddaemon/
 * and http://phpmultithreaddaemon.blogspot.com/ for more information
 *
 * Modifications by Daniel Kadosh, Affinegy Inc., June 2009
 * Made many enhancements for robustness:
 * - Truly daemonized by starting process group with posix_setsid()
 * - Surrounded key items in try/catch for proper logging
 * - Handling of SIGTERM/ SIGQUIT = no more children spawned, wait for all threads to die.
 * - Handling of SIGHUP = call to loadConfig() method
 * - Created PID file to ensure only 1 copy of the daemon is running
 * - Full cleanup of semaphores & PID file on exit
 *
 */
error_reporting(E_ALL);
require_once
'class.MTDaemon.php';

// Set number of threads for daemon
$nThreads = 2;

// Optional: Init logging class
$sLogfile = $argv[0].'.log'; // Set to null for logging to stdout
MTLog::getInstance($sLogfile)->setVerbosity(MTLog::INFO);

// Directory names
$sBaseDir = dirname(__FILE__);
$aDir['queue'] = "$sBaseDir/_queue"; // incoming files
$aDir['done'] = "$sBaseDir/_done"; // files parsed OK
$aDir['error'] = "$sBaseDir/_error"; // files not parsed OK

////////////////// For DEMO: populate _queue directory
$sQueuePath = $aDir['queue'];
if ( !
file_exists($sQueuePath) ) mkdir($sQueuePath, 0775, true);
if ( !
is_dir($sQueuePath) ) die('This should be a directory: '.$sQueuePath);
$sThisDir = dirname(__FILE__);
echo `
cp $sThisDir/*.php $sQueuePath`;
////////////////// For DEMO: populate _queue directory

class MTFileParser extends MTDaemon {
   
/*
     * Optional: read in config files or perform other "init" actions
     * Called from MTDaemon's _prerun(), plus when a SIGHUP signal is received.
     * NOTE: Runs under the PARENT process.
     */
   
public function loadConfig() {
       
// Already locked from within handle() loop, so don't lock/unlock.

        // Could load $aDir settings from a config file, but just using the global's setting here
       
global $aDir;
        foreach (
$aDir as $sPath ) {
            if ( !
file_exists($sPath) ) mkdir($sPath, 0775, true);
        }
       
$this->setVar('aDir', $aDir);

        global
$argv;
       
MTLog::getInstance()->info($argv[0].': Monitoring Queue dir='.$aDir['queue']);
    }

   
/*
     * Function to return quickly with (a) no work to do, or (b) next file to process
     * NOTE: Runs under the PARENT process
     * Could keep open DB connection to affinegy_master, and figure out $build here
     */
   
public function getNext($slot) {
       
$sFileToProcess = null;

       
// Get shared arrays
       
$this->lock();
       
$aDir = $this->getVar('aDir');
       
$aFileList = $this->getVar('aFileList'); // FIFO queue
       
$aFilesInProcess = $this->getVar('aFilesInProcess');
        if (
$aFilesInProcess==null ) $aFilesInProcess = array();

       
// Use ls to get files, only if the $aFileList queue is empty
       
if ( !$aFileList && is_dir($aDir['queue']) ) {
           
$sCmd = 'ls -1 '.$aDir['queue'];
           
exec($sCmd, $aFileList);
           
// Take out files already in process
           
if ( count($aFileList) && count($aFilesInProcess) ) {
               
$aFileList = array_diff($aFileList, $aFilesInProcess);
            }
        }

       
// Pull out a file to process from FIFO queue
       
if ( count($aFileList) ) {
           
$sFileToProcess = array_shift($aFileList);
           
array_push($aFilesInProcess, $sFileToProcess);
        }

       
// Store shared arrays back
       
$this->setVar('aFileList', $aFileList);
       
$this->setVar('aFilesInProcess', $aFilesInProcess);
       
$this->unlock();
        if ( !
$sFileToProcess ) return null;
        return
$aDir['queue'].'/'.$sFileToProcess; // $sFileName in run() method below
   
}

   
/*
     * Do main work here.
     * NOTE: Runs under a CHILD process
     */
   
public function run($sFileName, $slot) {
       
$sMsg = 'slot='.$slot.' file='.basename($sFileName);
       
MTLog::getInstance()->info('## Start '.$sMsg);
        try {
           
// Call parsing function that returns TRUE if OK, FALSE if not.
            // Could instantiate an object of a file-parsing class and run it instead...
           
$bProcessedOK = ParseFile($sFileName);
        } catch(
Exception $e ) {
           
MTLog::getInstance()->error('ProcessFile error '.$sMsg.': '.$e->getMessage());
           
$bProcessedOK = false;
        }

       
// Done, take file off "in-process" list
       
$this->lock();
       
$aFilesInProcess = $this->getVar('aFilesInProcess');
       
$nFileToRemove = array_search(basename($sFileName), $aFilesInProcess);
        unset(
$aFilesInProcess[$nFileToRemove]);
       
$this->setVar('aFilesInProcess', $aFilesInProcess);

       
// Move file out of queue directory
       
$aDir = $this->getVar('aDir');
       
$sDestDir = ($bProcessedOK ? $aDir['done'] : $aDir['error']);
       
$sDestFile = $sDestDir.'/'.basename($sFileName);
       
rename($sFileName, $sDestFile);
       
$this->unlock();

       
MTLog::getInstance()->info('-- End '.$sMsg);
        return
0;
    }
}

// For demo purposes, function that actually does the work on a file
function ParseFile($sFileName) {
   
$sContent = file_get_contents($sFileName);
    if ( !
$sContent ) return false;

   
$nRand = rand(10, 20);
   
MTLog::getInstance()->info('ParseFile() read '.strlen($sContent).' bytes from file '.$sFileName.' |pausing '.$nRand.' seconds');
   
sleep($nRand);
    return
true;
}

// Run daemon, start threads
try {
   
$mttest = null;
   
$mttest = new MTFileParser($nThreads); // Init
   
$mttest->handle(); // Run threads
} catch( Exception $e ) {
    if (
$mttest==null ) {
       
$sErr = $argv[0].': Daemon failed to start: '.$e->getMessage();
    } else {
       
$sErr = $argv[0].': Daemon died: '.$e->getMessage();
    }
   
MTLog::getInstance()->error($sErr);
    die(
$sErr."\n");
}

?>