Changes from pull request vyuldashev#94

Author: vyuldashev

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/RabbitMQQueue.php

@@ -3,6 +3,9 @@
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Queue;
 
 use DateTime;
+use ErrorException;
+use Exception;
+use Log;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue;
 use PhpAmqpLib\Channel\AMQPChannel;
@@ -30,6 +33,7 @@ class RabbitMQQueue extends Queue implements QueueContract
     protected $defaultQueue;
     protected $configQueue;
     protected $configExchange;
+    protected $sleepOnError;
 
     /**
      * @var int
@@ -53,6 +57,7 @@ public function __construct(AMQPStreamConnection $amqpConnection, $config)
         $this->configExchange = $config['exchange_params'];
         $this->declareExchange = $config['exchange_declare'];
         $this->declareBindQueue = $config['queue_declare_bind'];
+        $this->sleepOnError = isset($config['sleep_on_error']) ? $config['sleep_on_error'] : 5;
 
         $this->channel = $this->getChannel();
     }
@@ -83,32 +88,38 @@ 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']) && $options['delay'] > 0) {
-            list($queue, $exchange) = $this->declareDelayedQueue($queue, $options['delay']);
-        } else {
-            list($queue, $exchange) = $this->declareQueue($queue);
+        try {
+            $this->declareQueue($queue);
+            if (isset($options['delay']) && $options['delay'] > 0) {
+                list($queue, $exchange) = $this->declareDelayedQueue($queue, $options['delay']);
+            } else {
+                list($queue, $exchange) = $this->declareQueue($queue);
+            }
+
+            $headers = [
+                'Content-Type' => 'application/json',
+                'delivery_mode' => 2,
+            ];
+
+            if (isset($this->attempts) === true) {
+                $headers['application_headers'] = [self::ATTEMPT_COUNT_HEADERS_KEY => ['I', $this->attempts]];
+            }
+
+            // push job to a queue
+            $message = new AMQPMessage($payload, $headers);
+
+            $correlationId = $this->getCorrelationId();
+            $message->set('correlation_id', $correlationId);
+
+            // push task to a queue
+            $this->channel->basic_publish($message, $exchange, $queue);
+
+            return $correlationId;
+        } catch (ErrorException $exception) {
+            $this->reportConnectionError('pushRaw', $exception);
         }
 
-        $headers = [
-            'Content-Type' => 'application/json',
-            'delivery_mode' => 2,
-        ];
-
-        if (isset($this->attempts) === true) {
-            $headers['application_headers'] = [self::ATTEMPT_COUNT_HEADERS_KEY => ['I', $this->attempts]];
-        }
-
-        // push job to a queue
-        $message = new AMQPMessage($payload, $headers);
-
-        $correlationId = $this->getCorrelationId();
-        $message->set('correlation_id', $correlationId);
-
-        // push task to a queue
-        $this->channel->basic_publish($message, $exchange, $queue);
-
-        return $correlationId;
+        return null;
     }
 
     /**
@@ -137,15 +148,21 @@ 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 $exception) {
+            $this->reportConnectionError('pop', $exception);
         }
+
+        return null;
     }
 
     /**
@@ -175,7 +192,7 @@ private function declareQueue($name)
         $name = $this->getQueueName($name);
         $exchange = $this->configExchange['name'] ?: $name;
 
-        if ($this->declareExchange && ! in_array($exchange, $this->declaredExchanges)) {
+        if ($this->declareExchange && !in_array($exchange, $this->declaredExchanges)) {
             $this->declaredExchanges[] = $exchange;
             // declare exchange
             $this->channel->exchange_declare(
@@ -187,7 +204,7 @@ private function declareQueue($name)
             );
         }
 
-        if ($this->declareBindQueue && ! in_array($name, $this->declaredQueues)) {
+        if ($this->declareBindQueue && !in_array($name, $this->declaredQueues)) {
             $this->declaredQueues[] = $name;
             // declare queue
             $this->channel->queue_declare(
@@ -216,7 +233,7 @@ private function declareDelayedQueue($destination, $delay)
         $delay = $this->getSeconds($delay);
         $destination = $this->getQueueName($destination);
         $destinationExchange = $this->configExchange['name'] ?: $destination;
-        $name = $this->getQueueName($destination).'_deferred_'.$delay;
+        $name = $this->getQueueName($destination) . '_deferred_' . $delay;
         $exchange = $this->configExchange['name'] ?: $destination;
 
         // declare exchange
@@ -282,4 +299,16 @@ public function getCorrelationId()
     {
         return $this->correlationId ?: uniqid();
     }
+
+    /**
+     * @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);
+    }
+
 }