QueueConsumerAbstract.php — 10Web Booster – Website speed optimization, Cache & Page Speed optimizer trunk

tenweb-speed-optimizer / vendor / 10web-utils / 10web-queue / src / TenWebQueue / Abstracts / QueueConsumerAbstract.php

tenweb-speed-optimizer / vendor / 10web-utils / 10web-queue / src / TenWebQueue / Abstracts Last commit date

QueueConsumerAbstract.php

72 lines

1	<?php
2	namespace TenWebQueue\Abstracts;
3
4	use Interop\Queue\Consumer;
5	use Interop\Queue\Message;
6	use TenWebQueue\Interfaces\QueueConsumerInterface;
7	use TenWebQueue\QueueContext;
8
9	abstract class QueueConsumerAbstract implements QueueConsumerInterface
10	{
11
12	/**
13	* @var QueueContext
14	*/
15	private $queueContext;
16
17	/**
18	* @var Consumer
19	*/
20	private $consumer;
21
22	/**
23	* @param QueueContext $context
24	*/
25	public function __construct($context)
26	{
27	$this->queueContext = $context;
28	$this->consumer = $this->queueContext->getContext()->createConsumer($context->getQueue());
29	}
30
31	/**
32	* @param $requeue
33	*
34	* @return bool
35	*/
36	public function run($requeue = false)
37	{
38	$message = $this->consumer->receive(1000);
39	if ($message) {
40	if ($this->process($message)) {
41	$this->consumer->acknowledge($message);
42	} else {
43	$this->consumer->reject($message, $requeue);
44	}
45
46	return true;
47	}
48
49	return false;
50	}
51
52	/**
53	* @return string
54	*/
55	public function getQueueName()
56	{
57	return $this->queueContext->getQueue()->getQueueName();
58	}
59
60	/**
61	* @param Message $message
62	*
63	* @return boolean
64	*/
65	abstract public function process($message);
66
67
68	/**
69	* @return bool
70	*/
71	abstract public function finish();
72	}