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;
@@ -24,6 +27,8 @@ class RabbitMQQueue extends Queue implements QueueContract
     protected $declareExchange;
     protected $declaredExchanges = [];
     protected $declareBindQueue;
+    protected $sleepOnError;
+
     protected $declaredQueues = [];
 
     protected $defaultQueue;
@@ -52,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();
     }
@@ -92,33 +98,39 @@ 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);
-        }
-
-        $headers = [
-            'Content-Type' => 'application/json',
-            'delivery_mode' => 2,
-        ];
-
-        if (isset($this->attempts) === true) {
-            $headers['application_headers'] = [self::ATTEMPT_COUNT_HEADERS_KEY => ['I', $this->attempts]];
+        try {
+            $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);
+            }
+
+            $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);
         }
 
-        // 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;
     }
 
     /**
@@ -147,15 +159,22 @@ 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;
     }
 
     /**
@@ -299,4 +318,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);
+    }
+
 }

src/config/rabbitmq.php

@@ -12,8 +12,8 @@
     'host' => env('RABBITMQ_HOST', '127.0.0.1'),
     'port' => env('RABBITMQ_PORT', 5672),
 
-    'vhost'    => env('RABBITMQ_VHOST', '/'),
-    'login'    => env('RABBITMQ_LOGIN', 'guest'),
+    'vhost' => env('RABBITMQ_VHOST', '/'),
+    'login' => env('RABBITMQ_LOGIN', 'guest'),
     'password' => env('RABBITMQ_PASSWORD', 'guest'),
 
     'queue' => env('RABBITMQ_QUEUE'),
@@ -25,9 +25,9 @@
     // create the queue if not exists and bind to the exchange
 
     'queue_params' => [
-        'passive'     => env('RABBITMQ_QUEUE_PASSIVE', false),
-        'durable'     => env('RABBITMQ_QUEUE_DURABLE', true),
-        'exclusive'   => env('RABBITMQ_QUEUE_EXCLUSIVE', false),
+        'passive' => env('RABBITMQ_QUEUE_PASSIVE', false),
+        'durable' => env('RABBITMQ_QUEUE_DURABLE', true),
+        'exclusive' => env('RABBITMQ_QUEUE_EXCLUSIVE', false),
         'auto_delete' => env('RABBITMQ_QUEUE_AUTODELETE', false),
     ],
     'exchange_params' => [
@@ -40,4 +40,6 @@
         '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
+
 ];