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

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

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

Abstracts 4 years ago DTO 4 years ago Examples 4 years ago Exceptions 4 years ago Interfaces 4 years ago Traits 3 years ago QueueContext.php 4 years ago QueueOrchestrator.php 4 years ago

QueueContext.php

245 lines

1	<?php
2	namespace TenWebQueue;
3
4	use Enqueue\Fs\FsConnectionFactory;
5	use Interop\Queue\Context;
6	use Interop\Queue\Queue;
7	use TenWebQueue\DTO\QueueConfigDTO;
8	use TenWebQueue\DTO\QueueDataDTO;
9	use TenWebQueue\Exceptions\QueueException;
10	use TenWebQueue\Interfaces\QueueConsumerInterface;
11	use TenWebQueue\Interfaces\QueueProducerInterface;
12	use TenWebQueue\Traits\QueueRestart;
13
14
15	class QueueContext
16	{
17	use QueueRestart;
18
19	/**
20	* @var Context $context
21	*/
22	private $context = null;
23	/**
24	* @var Queue $queue
25	*/
26	private $queue;
27	/**
28	* @var QueueProducerInterface $producer
29	*/
30	private $producer;
31
32	/**
33	* @var QueueConsumerInterface $consumer
34	*/
35	private $consumer;
36
37	/**
38	* @var QueueConfigDTO $config
39	*/
40	private $config;
41
42	/**
43	* @var string[]
44	*/
45	private $connectionTypes = ['file'];
46
47
48	/**
49	* @param QueueConfigDTO $config
50	*
51	* @throws QueueException
52	*/
53	private function __construct($config)
54	{
55	$this->config = $config;
56	$this->setStartTime(microtime(true));
57	$this->purgeQueue();
58	$this->createContext();
59	}
60
61	/**
62	* @param $producerClass
63	* @param $consumerClass
64	*
65	* @return void
66	*/
67	public function setUpQueue($producerClass, $consumerClass)
68	{
69	$this->createQueue();
70
71	$this->createProducer($producerClass);
72	$this->createConsumer($consumerClass);
73	}
74
75	/**
76	* @return void
77	*/
78	public function unSetQueue()
79	{
80	$this->queue = null;
81	$this->producer = null;
82	$this->consumer = null;
83	}
84
85
86	/**
87	* @param QueueConfigDTO $config
88	* @param $producerClass
89	* @param $consumerClass
90	*
91	* @return QueueContext
92	* @throws QueueException
93	*/
94	public static function getInstance($config, $producerClass, $consumerClass)
95	{
96	$instance = self::readObjectFromFile($config->restartFilePath);
97	if (!($instance instanceof self) \|\| $config->force) {
98	$instance = new self($config);
99	}
100	$instance->setUpQueue($producerClass, $consumerClass);
101
102	return $instance;
103	}
104
105	/**
106	* @return void
107	* @throws QueueException
108	*/
109	public function createContext()
110	{
111	if (!in_array($this->config->connectionType, $this->connectionTypes)) {
112	throw new QueueException('Unknown connection type');
113	}
114	$connectionFactory = null;
115	if ($this->config->connectionType === 'file') {
116	if (!is_dir($this->config->fileConnectionPath)) {
117	throw new QueueException('Incorrect connection file path');
118	}
119	$connectionFactory = new FsConnectionFactory($this->config->fileConnectionPath);
120	}
121
122	if ($connectionFactory) {
123	$this->context = $connectionFactory->createContext();
124	}
125
126	}
127
128	/**
129	* @return void
130	*/
131	public function createQueue()
132	{
133	$this->queue = $this->context->createQueue($this->config->queueData->name);
134	}
135
136	/**
137	* @return void
138	*/
139	public function purgeQueue()
140	{
141	if (file_exists($this->config->fileConnectionPath . '/' . $this->config->queueData->name)) {
142	unlink($this->config->fileConnectionPath . '/' . $this->config->queueData->name);
143	}
144	if (file_exists($this->config->restartFilePath)) {
145	unlink($this->config->restartFilePath);
146	}
147	}
148
149	/**
150	* @param $producerClass
151	*
152	* @return void
153	*/
154	public function createProducer($producerClass)
155	{
156	$this->producer = new $producerClass($this);
157	}
158
159	/**
160	* @param $consumerClass
161	*
162	* @return void
163	*/
164	public function createConsumer($consumerClass)
165	{
166	$this->consumer = new $consumerClass($this);
167	}
168
169	/**
170	* @param array $data
171	*
172	* @return void
173	*/
174	public function enqueue($data)
175	{
176	if ($this->config->queueData->multipleItems) {
177	$this->producer->enqueueMany($data);
178	} else {
179	$this->producer->enqueueOne($data);
180	}
181	}
182
183	/**
184	* @return void
185	*/
186	public function dequeue($single = false)
187	{
188	if ($single) {
189	$this->consumer->run();
190	} else {
191	$count = 1;
192	while ($this->consumer->run()) {
193	if ($this->ifRestartNeeded($count, $this->config->itemsCountForRestart)) {
194	$this->unSetQueue();
195	self::writeObjectInfile($this, $this->config->restartFilePath);
196	$this->restart($this->config->restartRoute, $this->config->restartBody);
197	}
198	$count++;
199	}
200	$this->purgeQueue();
201	$this->consumer->finish();
202	}
203	}
204
205	/**
206	* @return Queue
207	*/
208	public function getQueue()
209	{
210	return $this->queue;
211	}
212
213	/**
214	* @return QueueProducerInterface
215	*/
216	public function getProducer()
217	{
218	return $this->producer;
219	}
220
221	/**
222	* @return QueueConsumerInterface
223	*/
224	public function getConsumer()
225	{
226	return $this->consumer;
227	}
228
229	/**
230	* @return QueueDataDTO
231	*/
232	public function getQueueData()
233	{
234	return $this->config->queueData;
235	}
236
237	/**
238	* @return Context
239	*/
240	public function getContext()
241	{
242	return $this->context;
243	}
244
245	}