Added better error handling to rabbitmq connectors

If the RabbitMQ connector was unavailable, and the queue was running in daemon mode, the error log would flood within seconds. This reports the connection error responsibly, and sleeps to stop the log file from being instantly flooded.

Author: samlev

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/RabbitMQQueue.php

@@ -3,8 +3,11 @@
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Queue;
 
 use DateTime;
+use ErrorException;
+use Exception;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue;
+use Log;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPConnection;
 use PhpAmqpLib\Message\AMQPMessage;
@@ -23,6 +26,7 @@ class RabbitMQQueue extends Queue implements QueueContract
 	protected $defaultQueue;
 	protected $configQueue;
 	protected $configExchange;
+	protected $sleepOnError;
 
 	/**
 	 * @param AMQPConnection $amqpConnection
@@ -36,6 +40,7 @@ public function __construct(AMQPConnection $amqpConnection, $config)
 		$this->configExchange = $config['exchange_params'];
 		$this->declareExchange = $config['exchange_declare'];
 		$this->declareBindQueue = $config['queue_declare_bind'];
+		$this->sleepOnError = $config['sleep_on_error'];
 
 		$this->channel = $this->getChannel();
 	}
@@ -66,20 +71,24 @@ public function push($job, $data = '', $queue = null)
 	public function pushRaw($payload, $queue = null, array $options = [])
 	{
 		$queue = $this->getQueueName($queue);
-		$this->declareQueue($queue);
-		if (isset($options['delay'])) {
-			$queue = $this->declareDelayedQueue($queue, $options['delay']);
+		try {
+			$this->declareQueue($queue);
+			if (isset($options['delay'])) {
+				$queue = $this->declareDelayedQueue($queue, $options['delay']);
+			}
+
+			// push job to a queue
+			$message = new AMQPMessage($payload, [
+				'Content-Type'  => 'application/json',
+				'delivery_mode' => 2,
+			]);
+
+			// push task to a queue
+			$this->channel->basic_publish($message, $queue, $queue);
+		} catch (ErrorException $e) {
+			$this->reportConnectionError('pushRaw', $e);
 		}
 
-		// push job to a queue
-		$message = new AMQPMessage($payload, [
-			'Content-Type'  => 'application/json',
-			'delivery_mode' => 2,
-		]);
-
-		// push task to a queue
-		$this->channel->basic_publish($message, $queue, $queue);
-
 		return true;
 	}
 
@@ -109,14 +118,18 @@ public function pop($queue = null)
 	{
 		$queue = $this->getQueueName($queue);
 
-		// declare queue if not exists
-		$this->declareQueue($queue);
+		try {
+			// declare queue if not exists
+			$this->declareQueue($queue);
 
-		// get envelope
-		$message = $this->channel->basic_get($queue);
+			// get envelope
+			$message = $this->channel->basic_get($queue);
 
-		if ($message instanceof AMQPMessage) {
-			return new RabbitMQJob($this->container, $this, $this->channel, $queue, $message);
+			if ($message instanceof AMQPMessage) {
+				return new RabbitMQJob($this->container, $this, $this->channel, $queue, $message);
+			}
+		} catch (ErrorException $e) {
+			$this->reportConnectionError('pop', $e);
 		}
 
 		return null;
@@ -215,4 +228,15 @@ private function declareDelayedQueue($destination, $delay)
 		return $name;
 	}
 
+	/**
+	 * @param string    $action
+	 * @param Exception $e
+	 */
+	private function reportConnectionError($action, Exception $e)
+	{
+		Log::error('AMQP error while attempting ' . $action . ': ' . $e->getMessage());
+
+		// Sleep so that we don't flood the log file
+		sleep($this->sleepOnError);
+	}
 }

src/config/rabbitmq.php

@@ -33,4 +33,5 @@
         'auto_delete' 		=> env('RABBITMQ_EXCHANGE_AUTODELETE', false),
     ],
 
+    'sleep_on_error'        => env('RABBITMQ_ERROR_SLEEP', 5), // the number of seconds to sleep if there's an error communicating with rabbitmq
 ];