Solr_QueueIndexAbstract.php

<?php

require_once dirname(__DIR__) . '/QueueProcessAbstract.php';

abstract class Solr_QueueIndexAbstract extends QueueProcessAbstract
{
    /** @var string  */
    private $core = 'default';

    public function __construct()
    {
        $this->_optionalParameters[] = array('param' => 'max_pull', 'description' => 'Maximum number of messages to pull at one time');
        parent::__construct();
    }

    /**
     * Allow any children to initialize the indexer with
     * any defaults before we begin processing
     */
    abstract protected function initSolrIndexer();

    /**
     * Returns the Exporter from Catalog\Solr\Exporter\*
     *
     * @return \Saatchi_Solr_Exporter
     */
    abstract protected function getExporter();

    /**
     * If data is missing for the exporter, enrich it here.
     * By default, return the incoming array.
     * @param array options
     * @return array
     */
     protected function enrich(array &$options)
     {
         return $options;
     }

    /**
     * @param bool $file
     */
    public function run($file = false)
    {
        $this->initSolrIndexer();
        parent::run($file);
    }

    /**
     * @return bool
     */
    protected function processData()
    {
        try {
            if (count($this->updateData) === 0) {
                return false;
            }

            // filter valid options
            $this->logMsg('Run import with "' . count($this->updateData) . '" records');
            $this->logMsg('items being worked on ' . print_r($this->updateData, true));
            $options = array_filter($this->updateData, array($this, 'isValidEvent'));

            $enrichedData = $this->enrich($options);

            if (!empty($enrichedData)) {
                $exporter = $this->getExporter();
                $exporter->setOptions($enrichedData);
                $solrManager = new Saatchi_Solr_Manager($this->core);
                $solrManager->addExporter($exporter);
                $solrManager->import();
            }
            $this->updateData = array();
            return true;
        } catch (Exception $e) {
            $this->logMsg(print_r($e, true));
            return false;
        }
    }

    /**
     * @param string $msg
     */
    public function logMsg($msg)
    {
        $msg = sprintf('%s - %s', get_class($this), $msg);
        parent::logMsg($msg);
    }

    /**
     * @param string $core
     */
    public function setSolrCore($core)
    {
        $this->core = $core;
    }
}