Merge pull request vyuldashev#27 from ivan1986/fix_data_not_array

Fix data not array

Author: vyuldashev

src/FintechFab/LaravelQueueRabbitMQ/Queue/Jobs/RabbitMQJob.php

@@ -1,5 +1,6 @@
 <?php namespace FintechFab\LaravelQueueRabbitMQ\Queue\Jobs;
 
+use FintechFab\LaravelQueueRabbitMQ\Queue\RabbitMQQueue;
 use Illuminate\Queue\Jobs\Job;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Message\AMQPMessage;
@@ -10,11 +11,13 @@ class RabbitMQJob extends Job
 
 	protected $channel;
 	protected $queue;
+	protected $connection;
 	protected $message;
 
-	public function __construct($container, AMQPChannel $channel, $queue, AMQPMessage $message)
+	public function __construct($container, RabbitMQQueue $connection, AMQPChannel $channel, $queue, AMQPMessage $message)
 	{
 		$this->container = $container;
+		$this->connection = $connection;
 		$this->channel = $channel;
 		$this->queue = $queue;
 		$this->message = $message;
@@ -79,17 +82,10 @@ public function release($delay = 0)
 		$attempts = $this->attempts();
 
 		// write attempts to body
-		$body['data']['attempts'] = $attempts + 1;
+		$body['attempts'] = $attempts + 1;
 
-		$job = $body['job'];
-		$data = $body['data'];
-
-		// push back to a queue
-		if ($delay > 0) {
-			Queue::later($delay, $job, $data, $this->getQueue());
-		} else {
-			Queue::push($job, $data, $this->getQueue());
-		}
+		$this->connection->pushRaw(json_encode($body), $this->getQueue(),
+			$delay > 0 ? [ 'delay' => $delay ] : []);
 	}
 
 	/**
@@ -101,7 +97,7 @@ public function attempts()
 	{
 		$body = json_decode($this->message->body, true);
 
-		return isset($body['data']['attempts']) ? $body['data']['attempts'] : 0;
+		return isset($body['attempts']) ? $body['attempts'] : 0;
 	}
 
 	/**

src/FintechFab/LaravelQueueRabbitMQ/Queue/RabbitMQQueue.php

@@ -44,69 +44,49 @@ public function __construct(AMQPConnection $amqpConnection, $config)
 	 */
 	public function push($job, $data = '', $queue = null)
 	{
-		$queue = $this->getQueueName($queue);
-		$payload = $this->createPayload($job, $data);
-		$this->declareQueue($queue);
-
-		// push job to a queue
-		$message = new AMQPMessage($payload, [
-			'Content-Type'  => 'application/json',
-			'delivery_mode' => 2,
-		]);
-
-		$this->channel->basic_publish($message, $queue, $queue);
-
-		return true;
+		return $this->pushRaw($this->createPayload($job, $data), $queue, []);
 	}
 
 	/**
-	 * Push a raw payload onto the queue.
+	 * Push a new job onto the queue after a delay.
 	 *
-	 * @param  string $payload
-	 * @param  string $queue
-	 * @param  array  $options
+	 * @param  \DateTime|int $delay
+	 * @param  string        $job
+	 * @param  mixed         $data
+	 * @param  string        $queue
 	 *
 	 * @return mixed
 	 */
-	public function pushRaw($payload, $queue = null, array $options = [])
+	public function later($delay, $job, $data = '', $queue = null)
 	{
-		$queue = $this->getQueueName($queue);
-		$this->declareQueue($queue);
-
-		// 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;
+		return $this->pushRaw($this->createPayload($job, $data), $queue, ['delay' => $delay]);
 	}
 
 	/**
-	 * Push a new job onto the queue after a delay.
+	 * Push a raw payload onto the queue.
 	 *
-	 * @param  \DateTime|int $delay
-	 * @param  string        $job
-	 * @param  mixed         $data
-	 * @param  string        $queue
+	 * @param  string $payload
+	 * @param  string $queue
+	 * @param  array  $options
 	 *
 	 * @return mixed
 	 */
-	public function later($delay, $job, $data = '', $queue = null)
+	public function pushRaw($payload, $queue = null, array $options = [])
 	{
-		$payload = $this->createPayload($job, $data);
+		$queue = $this->getQueueName($queue);
 		$this->declareQueue($queue);
-		$queue = $this->declareDelayedQueue($queue, $delay);
+		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);
 
 		return true;
@@ -130,7 +110,7 @@ public function pop($queue = null)
 		$message = $this->channel->basic_get($queue);
 
 		if ($message instanceof AMQPMessage) {
-			return new RabbitMQJob($this->container, $this->channel, $queue, $message);
+			return new RabbitMQJob($this->container, $this, $this->channel, $queue, $message);
 		}
 
 		return null;