umpirsky / Extraload

use Ko\AmqpBroker;

class QueuedExtractor implements ExtractorInterface

{

    private $extractor;

    private $broker;

    private $proucer;

    public function __construct(ExtractorIteratorInterface $extractor, AmqpBroker $broker, $proucer)

    {

        $this->extractor = $extractor;

        $this->broker = $broker;

        $this->proucer = $proucer;

    }

    public function extract()

    {

        while (null !== $extracted = $this->extractor->extract()) {

            $this->broker->getProducer($this->proucer)->publish(serialize($extracted));

        }

    }

}

use Ko\AmqpBroker;

class TransformerConsumer implements TransformerInterface

{

    private $transformer;

    private $broker;

    private $producer;

    public function __construct(TransformerInterface $transformer, AmqpBroker $broker, $producer)

    {

        $this->transformer = $transformer;

        $this->broker = $broker;

        $this->producer = $producer;

    }

    public function transform($data = null)

    {

        $this->broker->getProducer($this->producer)->publish(serialize(

            $this->transformer->transform($data)

        ));

    }

}