Added IProducer + fixed some issues

src/Client.php

@@ -5,7 +5,7 @@
 namespace Contributte\RabbitMQ;
 
 use Contributte\RabbitMQ\Producer\Exception\ProducerFactoryException;
-use Contributte\RabbitMQ\Producer\Producer;
+use Contributte\RabbitMQ\Producer\IProducer;
 use Contributte\RabbitMQ\Producer\ProducerFactory;
 
 /**
@@ -23,7 +23,7 @@ public function __construct(private ProducerFactory $producerFactory)
 	/**
 	 * @throws ProducerFactoryException
 	 */
-	public function getProducer(string $name): Producer
+	public function getProducer(string $name): IProducer
 	{
 		return $this->producerFactory->getProducer($name);
 	}

src/Connection/Client.php

@@ -42,7 +42,7 @@ public function sendHeartbeat(): void
 		$this->getWriter()->appendFrame(new Protocol\HeartbeatFrame(), $this->writeBuffer);
 		$this->flushWriteBuffer();
 
-		$this->options['heartbeat_callback']?->call($this);
+		$this->options['heartbeat_callback'] && $this->options['heartbeat_callback']();
 	}
 
 	public function syncDisconnect(int $replyCode = 0, string $replyText = ""): bool
@@ -110,7 +110,7 @@ public function run($maxSeconds = null): void
 		}
 
 		do {
-			$this->options['cycle_callback']?->call($this);
+			$this->options['cycle_callback'] && $this->options['cycle_callback']();
 			if (!empty($this->queue)) {
 				$frame = array_shift($this->queue);
 			} else {

src/Connection/Connection.php

@@ -7,6 +7,9 @@
 use Bunny\Channel;
 use Bunny\Exception\ClientException;
 use Contributte\RabbitMQ\Connection\Exception\ConnectionException;
+use function in_array;
+use function max;
+use function time;
 
 final class Connection implements IConnection
 {
@@ -65,6 +68,7 @@ public function __construct(
 		$this->heartbeat = max($heartbeat, self::HEARTBEAT_INTERVAL);
 
 		if (!$lazy) {
+			$this->lastBeat = time();
 			$this->bunnyClient->connect();
 		}
 	}
@@ -134,6 +138,7 @@ public function connectIfNeeded(): void
 			return;
 		}
 
+		$this->lastBeat = time();
 		$this->bunnyClient->connect();
 	}
 

src/Consumer/Consumer.php

@@ -33,13 +33,11 @@ public function __construct(
 		$this->callback = $callback;
 	}
 
-
 	public function getQueue(): IQueue
 	{
 		return $this->queue;
 	}
 
-
 	public function getCallback(): callable
 	{
 		return $this->callback;
@@ -82,50 +80,24 @@ public function consume(?int $maxSeconds = null, ?int $maxMessages = null): void
 
 	protected function sendResponse(Message $message, Channel $channel, int $result, Client $client): void
 	{
-		switch ($result) {
-			case IConsumer::MESSAGE_ACK:
-				// Acknowledge message
-				$channel->ack($message);
-
-				break;
-
-			case IConsumer::MESSAGE_NACK:
-				// Message will be requeued
-				$channel->nack($message);
-
-				break;
-
-			case IConsumer::MESSAGE_REJECT:
-				// Message will be discarded
-				$channel->reject($message, false);
-
-				break;
-
-			case IConsumer::MESSAGE_REJECT_AND_TERMINATE:
-				// Message will be discarded
-				$channel->reject($message, false);
-				$client->stop();
-
-				break;
-
-			case IConsumer::MESSAGE_ACK_AND_TERMINATE:
-				// Acknowledge message and terminate
-				$channel->ack($message);
-				$client->stop();
-
-				break;
-
-			case IConsumer::MESSAGE_NACK_AND_TERMINATE:
-				// Message will be requeued
-				$channel->nack($message);
-				$client->stop();
-
-				break;
+		match ($result) {
+			IConsumer::MESSAGE_ACK, IConsumer::MESSAGE_ACK_AND_TERMINATE => $channel->ack($message),
+			IConsumer::MESSAGE_NACK, IConsumer::MESSAGE_NACK_AND_TERMINATE => $channel->nack($message),
+			IConsumer::MESSAGE_REJECT, IConsumer::MESSAGE_REJECT_AND_TERMINATE => $channel->reject($message, false),
+			default => throw new \InvalidArgumentException("Unknown return value of consumer [{$this->name}] user callback"),
+		};
 
-			default:
-				throw new \InvalidArgumentException(
-					"Unknown return value of consumer [{$this->name}] user callback"
-				);
+		if (in_array(
+			$result,
+			[
+				IConsumer::MESSAGE_REJECT_AND_TERMINATE,
+				IConsumer::MESSAGE_ACK_AND_TERMINATE,
+				IConsumer::MESSAGE_NACK_AND_TERMINATE
+			],
+			true
+		)
+		) {
+			$client->stop();
 		}
 	}
 

src/Diagnostics/BarPanel.php

@@ -4,7 +4,7 @@
 
 namespace Contributte\RabbitMQ\Diagnostics;
 
-use Contributte\RabbitMQ\Producer\Producer;
+use Contributte\RabbitMQ\Producer\IProducer;
 use Contributte\RabbitMQ\Producer\ProducerFactory;
 use Nette\Utils\Html;
 use Tracy\IBarPanel;
@@ -26,7 +26,7 @@ class BarPanel implements IBarPanel
 	public function __construct(private ProducerFactory $producerFactory)
 	{
 		$this->producerFactory->addOnCreatedCallback(
-			function (string $name, Producer $producer): void {
+			function (string $name, IProducer $producer): void {
 				$this->sentMessages[$name] = [];
 				$producer->addOnPublishCallback(
 					function (string $message) use ($name): void {

src/Producer/IProducer.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Contributte\RabbitMQ\Producer;
+
+interface IProducer
+{
+	public const DELIVERY_MODE_NON_PERSISTENT = 1;
+	public const DELIVERY_MODE_PERSISTENT = 2;
+
+	/**
+	 * @param array<string, string|int> $headers
+	 */
+	public function publish(string $message, array $headers = [], ?string $routingKey = null): void;
+	public function addOnPublishCallback(callable $callback): void;
+}

src/Producer/Producer.php

@@ -12,12 +12,8 @@
 use Contributte\RabbitMQ\LazyDeclarator;
 use Contributte\RabbitMQ\Queue\IQueue;
 
-final class Producer
+final class Producer implements IProducer
 {
-
-	public const DELIVERY_MODE_NON_PERSISTENT = 1;
-	public const DELIVERY_MODE_PERSISTENT = 2;
-
 	/**
 	 * @var callable[]
 	 */

src/Producer/ProducerFactory.php

@@ -18,20 +18,22 @@ final class ProducerFactory
 	public array $createdCallbacks = [];
 
 	/**
-	 * @var Producer[]
+	 * @var IProducer[]
 	 */
 	private array $producers = [];
 
-
-	public function __construct(private ProducersDataBag $producersDataBag, private QueueFactory $queueFactory, private ExchangeFactory $exchangeFactory, private LazyDeclarator $lazyDeclarator)
-	{
+	public function __construct(
+		private ProducersDataBag $producersDataBag,
+		private QueueFactory $queueFactory,
+		private ExchangeFactory $exchangeFactory,
+		private LazyDeclarator $lazyDeclarator
+	) {
 	}
 
-
 	/**
 	 * @throws ProducerFactoryException
 	 */
-	public function getProducer(string $name): Producer
+	public function getProducer(string $name): IProducer
 	{
 		if (!isset($this->producers[$name])) {
 			$this->producers[$name] = $this->create($name);
@@ -40,17 +42,15 @@ public function getProducer(string $name): Producer
 		return $this->producers[$name];
 	}
 
-
 	public function addOnCreatedCallback(callable $callback): void
 	{
 		$this->createdCallbacks[] = $callback;
 	}
 
-
 	/**
 	 * @throws ProducerFactoryException
 	 */
-	private function create(string $name): Producer
+	private function create(string $name): IProducer
 	{
 		try {
 			$producerData = $this->producersDataBag->getDataByKey($name);

src/Producer/ProducersDataBag.php

@@ -29,7 +29,7 @@ public function __construct(array $data)
 	 */
 	public function addProducerByData(string $producerName, array $data): void
 	{
-		$data['deliveryMode'] ??= Producer::DELIVERY_MODE_PERSISTENT;
+		$data['deliveryMode'] ??= IProducer::DELIVERY_MODE_PERSISTENT;
 		$data['contentType'] ??= 'text/plain';
 		$data['exchange'] ??= null;
 		$data['queue'] ??= null;