root/trunk/midcom/midcom.helper.replicator/queuemanager.php

<?php
/**
* @package midcom.helper.replicator
* @author The Midgard Project, http://www.midgard-project.org
* @version $Id: viewer.php 3975 2006-09-06 17:36:03Z bergie $
* @copyright The Midgard Project, http://www.midgard-project.org
* @license http://www.gnu.org/licenses/lgpl.html GNU Lesser General Public License
*/

/**
 * @package midcom.helper.replicator
 */
class midcom_helper_replicator_queuemanager extends midcom_baseclasses_components_purecode
{
    var $max_queues_per_pass = 10; // per subscription
    var $max_items_per_pass = 3500; // per QM instance, not this is checked only *after* queue pass so more might be processed if single queue dirs are very large
    var $_items_processed = 0;
    var $exporters = array();
    var $transporters = array();
    var $started = 0;
    /**
     * file counts keyed by subscription guid
     */
    var $file_counts = array();

    /**
     * Initializes the class. The real startup is done by the initialize() call.
     *
     * @param midcom_helper_replicator_subscription_dba $subscription Subscription
     */
    function midcom_helper_replicator_queuemanager()
    {
         $this->_component = 'midcom.helper.replicator';
         parent::midcom_baseclasses_components_purecode();
         $this->started = date('YmdHis');
    }

    /**
     * The correct way to get a queue manager, call this statically, returns reference
     * 
     * @static
     */
    function &get()
    {
        if (!array_key_exists('midcom_helper_replicator_queuemanager_instance', $GLOBALS))
        {
            $GLOBALS['midcom_helper_replicator_queuemanager_instance'] = new midcom_helper_replicator_queuemanager();
        }
        return $GLOBALS['midcom_helper_replicator_queuemanager_instance'];
    }

    function sanity_check()
    {
        // TODO: Sanity check queue root etc
        return true;
    }

    /**
     * Check if we have any valid queues to add data to
     *
     * @return boolean indicating state
     */
    function can_add_to_queue()
    {
        $global_base = $this->_config->get('queue_root_dir');
        if (   !is_dir($global_base)
            || !is_writable($global_base))
        {
            return false;
        }
        $qb = midcom_helper_replicator_subscription_dba::new_query_builder();
        $qb->add_constraint('status', '=', MIDCOM_REPLICATOR_AUTOMATIC);
        $qb->set_limit(1);
        $_MIDCOM->auth->request_sudo('midcom.helper.replicator');
        $count = $qb->count();
        $_MIDCOM->auth->drop_sudo();
        if ($count > 0)
        {
            return true;
        }
        return false;
    }

    /**
     * This method will check exporters of each subscription whether they are interested
     * in exporting the particular object.
     *
     * If the exporters are interested, they will be asked to serialize the data. Queue manager
     * will then store the serialized data for each queue.
     * @todo refactor to smaller methods
     * @todo query for subscriptions only once
     * @todo raise UIMessage in style of 'N object queued to subscription X'
     */
    function add_to_queue(&$object, $rewrite_to_delete = false)
    {
        $_MIDCOM->auth->request_sudo('midcom.helper.replicator');
        $GLOBALS['midcom_helper_replicator_logger']->push_prefix('Queue Manager');
        debug_push_class(__CLASS__, __FUNCTION__);
        $qb = midcom_helper_replicator_subscription_dba::new_query_builder();
        $qb->add_constraint('sitegroup', '=', $object->sitegroup);
        // NOTE: if this constraint is changed see can_add_to_queue()
        $qb->add_constraint('status', '=', MIDCOM_REPLICATOR_AUTOMATIC);
        $subscriptions = $qb->execute();
        $queued_guids = array();
        foreach ($subscriptions as $subscription)
        {
            debug_add("Processing subscription '{$subscription->title}' ({$subscription->guid})");
        
            $exporter = midcom_helper_replicator_exporter::create($subscription);

            if (!$exporter->is_exportable($object))
            {
                debug_add('exporter->is_exportable() returned false');
                continue;
            }
            
            if ($rewrite_to_delete)
            {
                $exporter->_serialize_rewrite_to_delete[$object->guid] = true;
            }
            $exporter_serializations = $exporter->serialize($object);
            if ($exporter_serializations === false)
            {
                debug_add('exporter->serialize() returned false', MIDCOM_LOG_ERROR);
                continue;
            }
            if (empty($exporter_serializations))
            {
                debug_add('exporter->serialize() returned empty array', MIDCOM_LOG_WARN);
                continue;
            }
            
            // Store the serialized XML data for the queue
            $path = $this->_get_subscription_queue_basedir($subscription);
            if (empty($path))
            {
                // TODO: error handling
                debug_add('could not get queue dir for subscription', MIDCOM_LOG_ERROR);
                continue;
            }
            if (   !isset($this->file_counts[$subscription->guid])
                || !is_numeric($this->file_counts[$subscription->guid]))
            {
                $this->file_counts[$subscription->guid] = 1;
            }
            $i =& $this->file_counts[$subscription->guid];
            debug_add('about to queue ' . count($exporter_serializations) . ' keys');
            reset($exporter_serializations);
            foreach ($exporter_serializations as $key => $data)
            {
                if (empty($data))
                {
                    $msg = "Key {$key} has empty data, skipping";
                    $GLOBALS['midcom_helper_replicator_logger']->log($msg, MIDCOM_LOG_WARN);
                    debug_add($msg, MIDCOM_LOG_WARN);
                    continue;
                }
                $file = "{$path}/" . sprintf('%010d', $i) . "_{$key}.xml";
                if (file_exists($file))
                {
                    // PANIC: file is already there (it *really* should not be)
                    unset($exporter_serializations, $key, $data);
                    debug_add("file {$file} already exists", MIDCOM_LOG_ERROR);
                    debug_pop();
                    $GLOBALS['midcom_helper_replicator_logger']->pop_prefix();
                    $_MIDCOM->auth->drop_sudo();
                    return false;
                }
                $fp = fopen($file, 'w');
                if (!$fp)
                {
                    // PANIC: Can't open file for writing
                    debug_add("can't open file {$file} for writing", MIDCOM_LOG_ERROR);
                    unset($exporter_serializations, $key, $data);
                    debug_pop();
                    $GLOBALS['midcom_helper_replicator_logger']->pop_prefix();
                    $_MIDCOM->auth->drop_sudo();
                    return false;
                }
                fwrite($fp, $data, strlen($data));
                fclose($fp);
                
                $msg = "Queued {$key} as {$file}";
                $GLOBALS['midcom_helper_replicator_logger']->log($msg);
                debug_add($msg);

                // TODO: How to call midgard_replicator::export() for all the objects exported ?? (and is this the correct place for that ?)
                if (($guid_end = strpos($key, '_metadata')) !== false)
                {
                    // This key is attachment metadata
                    $export_guid = substr($key, 0, $guid_end);
                    unset($guid_end);
                }
                elseif (mgd_is_guid($key))
                {
                    $export_guid = $key;
                }
                else
                {
                    debug_add("could not determine GUID from key '{$key}'");
                    $export_guid = false;
                }
                if ($export_guid)
                {
                    $queued_guids[$export_guid] = true;
                }

                unset($key, $data);
                ++$i;
            }
            unset($exporter_serializations);
            debug_add('all keys queued, now marking them exported');
            foreach ($queued_guids as $export_guid => $bool)
            {
                $marked_exported = false;
                if (version_compare(mgd_version(), '1.8.2', '>='))
                {
                    // In Midgard 1.8.2 we can do this efficiently with the export_by_guid method
                    $marked_exported = midgard_replicator::export_by_guid($export_guid);
                }
                else
                {
                    $object = $_MIDCOM->dbfactory->get_object_by_guid($export_guid);
                    $marked_exported = midgard_replicator::export($object);
                }
                
                if ($marked_exported)
                {
                    $msg = "Marked GUID '{$export_guid}' as exported to queue \"{$subscription->title}\"";
                    //$GLOBALS['midcom_helper_replicator_logger']->log($msg);
                    debug_add($msg);
                }
                else
                {
                    $msg = "Failed to mark GUID '{$export_guid}' as exported, errstr: " . mgd_errstr();
                    $GLOBALS['midcom_helper_replicator_logger']->log($msg, MIDCOM_LOG_ERROR);
                    debug_add($msg, MIDCOM_LOG_ERROR);
                }
            }
            debug_add("All done for subscription '{$subscription->title}' ({$subscription->guid})");
        }
        debug_add('All automatic subscriptions processed');
        debug_pop();
        $GLOBALS['midcom_helper_replicator_logger']->pop_prefix();
        $_MIDCOM->auth->drop_sudo();
        return true;
    }

    function list_path_items($path)
    {
        $path = preg_replace('%/{2,}|/$%', '', $path);
        $ret = array();
        $dp = opendir($path);
        if (!$dp)
        {
            return false;
        }
        
        while (($file_name = readdir($dp)) !== false)
        {
            if (   $file_name == '.'
                || $file_name == '..')
            {
                continue;
            }
            $file_path = "{$path}/{$file_name}";
            if (is_dir($file_path))
            {
                $ret = array_merge($ret, midcom_helper_replicator_queuemanager::list_path_items($file_path));
                continue;
            }
            $ret[] = $file_path;
        }
        closedir($dp);

        sort($ret);
        return $ret;
    }

    function get_sg_basedir(&$subscription)
    {
        // Normalize basedir, no trailing slash and no consecutive slashes
        $global_base = preg_replace('%/{2,}|/$%', '', $this->_config->get('queue_root_dir'));
        if (!is_dir($global_base))
        {
            // The configuration key might have dynamic part to it
            debug_push_class(__CLASS__, __FUNCTION__);    
            debug_add("directory {$global_base} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($global_base))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$global_base}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        // Append sitegroup name
        $sitegroup_base = "{$global_base}/" . $this->safe_sg_name($subscription->sitegroup);
        if (!is_dir($sitegroup_base))
        {
            // The configuration key might have dynamic part to it
            debug_push_class(__CLASS__, __FUNCTION__);    
            debug_add("directory {$sitegroup_base} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($sitegroup_base))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$sitegroup_base}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        return $sitegroup_base;
    }

    /**
     * Gets/creates the path for subscriptions spool dir
     * @todo make a smarter recursive directory creator
     */
    function get_subscription_basedir(&$subscription)
    {
        $sitegroup_base = $this->get_sg_basedir($subscription);
        if ($sitegroup_base === false)
        {
            return false;
        }

        $subscription_path = "{$sitegroup_base}/{$subscription->guid}";
        if (!is_dir($subscription_path))
        {
            debug_push_class(__CLASS__, __FUNCTION__);    
            debug_add("directory {$subscription_path} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($subscription_path))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$subscription_path}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        return $subscription_path;
    }

    function get_subscription_quarantine_basedir(&$subscription)
    {
        $sitegroup_base = $this->get_sg_basedir($subscription);
        if ($sitegroup_base === false)
        {
            return false;
        }

        $subscription_path = "{$sitegroup_base}/{$subscription->guid}-quarantine";
        if (!is_dir($subscription_path))
        {
            debug_push_class(__CLASS__, __FUNCTION__);    
            debug_add("directory {$subscription_path} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($subscription_path))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$subscription_path}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        return $subscription_path;
    }

    function _get_subscription_quarantine_queuedir(&$subscription)
    {
        $quarantine_path = $this->get_subscription_quarantine_basedir($subscription);
        if ($quarantine_path === false)
        {
            // TODO: Error reporting
            return false;
        }
        $queue_path = "{$quarantine_path}/{$this->started}";
        if (!is_dir($queue_path))
        {
            debug_push_class(__CLASS__, __FUNCTION__);
            debug_add("directory {$queue_path} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($queue_path))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$queue_path}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        return $queue_path;
    }


    function _get_subscription_queue_basedir(&$subscription)
    {
        $subscription_path = $this->get_subscription_basedir($subscription);
        if ($subscription_path === false)
        {
            // TODO: Error reporting
            return false;
        }
        $queue_path = "{$subscription_path}/{$this->started}";
        if (!is_dir($queue_path))
        {
            debug_push_class(__CLASS__, __FUNCTION__);
            debug_add("directory {$queue_path} does not exist, creating", MIDCOM_LOG_DEBUG);
            if (!mkdir($queue_path))
            {
                // TODO: Error reporting
                debug_add("could not create directory {$queue_path}", MIDCOM_LOG_ERROR);
                debug_pop();
                return false;
            }
            debug_pop();
        }
        return $queue_path;
    }

    function _process_queue_queuepath_sanitychecks(&$queue_name, &$subscription_path)
    {
        if (   $queue_name == '.'
            || $queue_name == '..')
        {
            // Skip the . and .. entries (which are always present)
            return false;
        }
        // Sanity check the subdir name
        if (!is_numeric($queue_name))
        {
            // Nonnumeric paths are not our queues
            debug_add("Weird queue name '{$queue_name}' in path '{$subscription_path}'", MIDCOM_LOG_WARN);
            return false;
        }
        // Sanity check path
        $queue_path = "{$subscription_path}/{$queue_name}";
        if (!is_dir($queue_path))
        {
            debug_add("Queue path '{$queue_path}' is not a directory, skipping", MIDCOM_LOG_ERROR);
            return false;
        }
        return true;
    }

    /**
     * Helper for process_queue, moves quarantined items to correct directory
     */
    function _quarantine_items(&$q_items, &$items_paths, &$subscription)
    {
        if (!is_array($q_items))
        {
            return false;
        }
        if (empty($q_items))
        {
            return true;
        }
        $quarantine_path = $this->_get_subscription_quarantine_queuedir($subscription);
        if (!is_dir($quarantine_path))
        {
            // Could not get valid dir
            return false;
        }
        foreach ($q_items as $item_key => $item_path)
        {
            // Reset time limit counter while processing files
            set_time_limit(30);
            $quarantine_filepath = $quarantine_path . '/' . basename($item_path);
            debug_add("Quarantineing '{$item_key}' as '{$quarantine_filepath}'", MIDCOM_LOG_DEBUG);
            $output = array();
            $code = 0;
            exec("mv {$item_path} {$quarantine_filepath}", $output, $code);
            if ($code != 0)
            {
                debug_add("Failed to quarantine '{$item_path}' as '{$quarantine_filepath}'", MIDCOM_LOG_ERROR);
                continue;
            }

            // TODO: write per-item error if available to some index file.

            unset($q_items[$item_key], $items_paths[$item_key]);
        }
        return true;
    }

    /**
     * Helper for process_queue, removes processed items, quarantines failed ones
     */
    function _process_queue_quarantines(&$items, &$items_paths, &$subscription)
    {
        debug_push_class(__CLASS__, __FUNCTION__);
        $q_items = array();
        foreach ($items_paths as $item_key => $item_path)
        {
            if (array_key_exists($item_key, $items))
            {
                // Item still in array (transporter should remove each key as it's transported properly)
                debug_add("Transporter left key '{$item_key}' into items, marking for quarantineing", MIDCOM_LOG_INFO);
                $q_items[$item_key] = $item_path;
                continue;
            }
            if (!unlink($item_path))
            {
                debug_add("Could not remove file '{$item_path}'", MIDCOM_LOG_ERROR);
                continue;
            }
            //$GLOBALS['midcom_helper_replicator_logger']->log("File {$item_path} removed from queue \"{$subscription->title}\"");
            unset($items_paths[$item_key]);
        }
        $this->_quarantine_items(&$q_items, &$items_paths, &$subscription);
        unset($q_items);
        // Restore to original value
        set_time_limit(ini_get('max_execution_time'));
        debug_pop();
    }