BufferedQueue.php

<?php

namespace Rohlikcz\RabbitMq;

use Rohlikcz\Identified;
use Rohlikcz\InvalidArgumentException;
use Rohlikcz\InvalidStateException;
use Doctrine\Common\Util\ClassUtils;
use Kdyby;
use Nette;
use Nette\Utils\Json;
use PhpAmqpLib\Message\AMQPMessage;



abstract class BufferedQueue extends Nette\Object implements Kdyby\RabbitMq\IConsumer, Kdyby\Events\Subscriber
{

	/**
	 * @var array
	 */
	protected $dirtyEntities = [];

	/**
	 * @var Kdyby\RabbitMq\Connection
	 */
	protected $rabbitmq;

	/**
	 * @var string
	 */
	protected $exchangeName;



	public function __construct($exchangeName, Kdyby\RabbitMq\Connection $rabbitmq)
	{
		$this->rabbitmq = $rabbitmq;
		$this->exchangeName = $exchangeName;
	}



	public function getSubscribedEvents()
	{
		return ['Nette\Application\Application::onShutdown' => 'flushSync'];
	}



	protected function markDirty(Identified $entity)
	{
		if (empty($entity->getId())) {
			throw new InvalidArgumentException("Entity has no identifier");
		}

		$class = ClassUtils::getRealClass(get_class($entity));
		$this->dirtyEntities[$class][$entity->getId()] = TRUE;
	}



	protected function prepare($type, $id)
	{
		return Json::encode(['type' => $type, 'id' => $id]);
	}



	public function flushSync()
	{
		if ($this->exchangeName === NULL) {
			throw new InvalidStateException('Exchange name is not defined');
		}

		if (!array_filter($this->dirtyEntities)) {
			return;
		}

		$producer = $this->rabbitmq->getProducer($this->exchangeName);

		foreach ($this->dirtyEntities as $type => $entities) {
			foreach ($entities as $id => $_) {
				$producer->publish($this->prepare($type, $id));
			}
		}

		$this->dirtyEntities = [];
	}



	abstract public function process(AMQPMessage $message);

}