for testing and deploying your application
for finding and fixing issues
for empowering human code reviews
<?php
namespace Darkilliant\MqProcessBundle\Message\Publisher;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
class AmqpMessagePublisher
{
/** @var AMQPChannel */
private $channel;
/** @var string */
private $queue;
private $exchange;
/** @var int */
private $batch = 0;
/** @var bool */
private $persistant;
public function __construct(AMQPChannel $channel, string $queue, string $exchange, bool $persistant)
$this->channel = $channel;
$this->queue = $queue;
$this->exchange = $exchange;
$this->persistant = $persistant;
$this->channel->queue_declare($this->queue, false, $this->persistant);
$this->channel->queue_bind($this->queue, $this->exchange, $this->queue);
}
public function publish(string $message, int $batchCount)
++$this->batch;
$this->channel->batch_basic_publish(new AMQPMessage($message), $this->exchange, $this->queue);
if ($this->batch >= $batchCount) {
$this->flush();
public function finalize()
public function flush()
$this->channel->publish_batch();
public function countMessages(): int
return $this->channel->queue_declare($this->queue, true)->messageCount;
jean-pasqualini /
import
