Worker

<?php

namespace Mailbroker\MailDetailsBundle\Worker;

use Mmoreram\GearmanBundle\Driver\Gearman;
use Doctrine\ORM\EntityManager;
use GearmanJob;
use Mmoreram\GearmanBundle\Service\GearmanClient;

use Mailbroker\MailDetailsBundle\Importer\ImporterInterface;
use Mailbroker\MailDetailsBundle\Canoniser\CanoniserFactory;

use Mailbroker\MailDetailsBundle\Entity\ConverterReport;

/**
 * PendingUpload worker for data import via canoniser
 *
 * @Gearman\Work(
 *     description = "Converter worker",
 *     service = "mailbroker_mail_details.worker.converter"
 * )
 */
class Converter
{
    protected $em;
    protected $importer;
    protected $canoniserFactory;
    protected $gearman;
    protected $canoniserSettings;

    /**
     * Inject dependencies and create the object
     * 
     * @param EntityManager        $em
     * @param ImporterInterface    $importer
     * @param CanoniserFactory     $canoniserFactory
     * @param GearmanClient        $gearman
     * @param array                $canoniserSettings
     */
    public function __construct(EntityManager $em, ImporterInterface $importer, CanoniserFactory $canoniserFactory, GearmanClient $gearman, array $canoniserSettings)
    {
        $this->em = $em;
        $this->importer = $importer;
        $this->canoniserFactory = $canoniserFactory;
        $this->gearman = $gearman;
        $this->canoniserSettings = $canoniserSettings;
    }

    /**
     * Import from source
     * 
     * @param GearmanJob $job
     * 
     * @return boolean
     *
     * @Gearman\Job(
     *     name = "import"
     * )
     */
    public function import(GearmanJob $job)
    {
        $input = json_decode($job->workload(), true);

        $importer = $this->importer;
        $importer->importFile($input['path']);
        $importer->setMappedKeys($input['mapping']);

        $data = $importer->getData();

        $this->gearman->addTaskBackground('MailbrokerMailDetailsBundleWorkerConverter~convert', json_encode([
            'rows' => $data,
            'mapping' => $input['mapping'],
            'start' => 0,
            'end' => count($data),
        ]));

        $this->gearman->runTasks();

        return true;
    }

    /**
     * Convert imported rows
     *     
     * @param GearmanJob $job
     * 
     * @return boolean
     * 
     * @Gearman\Job(
     *     name = "convert"
     * )
     */
    public function convert(GearmanJob $job)
    {
        extract(json_decode($job->workload(), true));

        $em = $this->em;
        $repo = $em->getRepository('MailbrokerMailDetailsBundle:ConverterReport');

        $report = new ConverterReport();
        $report->setStart($start);
        $report->setEnd($end);
        
        $jobSize = count($rows);

        foreach ($rows as $i => $row) {
            $job->sendStatus($i, $jobSize);
            echo "Hi!\n";
            // sleep(3);
        }

        $job->sendComplete('Done!');

        $report->setFinishedAt(new \DateTime());
        $em->persist($report);
        $em->flush();

        return true;
    }
}