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

tenweb-speed-optimizer / vendor / enqueue / fs / FsContext.php

tenweb-speed-optimizer / vendor / enqueue / fs Last commit date

.github 2 years ago CannotObtainLockException.php 4 years ago FsConnectionFactory.php 4 years ago FsConsumer.php 4 years ago FsContext.php 4 years ago FsDestination.php 4 years ago FsMessage.php 4 years ago FsProducer.php 4 years ago LICENSE 4 years ago LegacyFilesystemLock.php 4 years ago Lock.php 4 years ago README.md 2 years ago composer.json 2 years ago

FsContext.php

206 lines

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Enqueue\Fs;
6
7	use Interop\Queue\Consumer;
8	use Interop\Queue\Context;
9	use Interop\Queue\Destination;
10	use Interop\Queue\Exception\InvalidDestinationException;
11	use Interop\Queue\Exception\SubscriptionConsumerNotSupportedException;
12	use Interop\Queue\Message;
13	use Interop\Queue\Producer;
14	use Interop\Queue\Queue;
15	use Interop\Queue\SubscriptionConsumer;
16	use Interop\Queue\Topic;
17	use Makasim\File\TempFile;
18	use Symfony\Component\Filesystem\Filesystem;
19
20	class FsContext implements Context
21	{
22	/**
23	* @var string
24	*/
25	private $storeDir;
26
27	/**
28	* @var int
29	*/
30	private $preFetchCount;
31
32	/**
33	* @var int
34	*/
35	private $chmod;
36
37	/**
38	* @var int
39	*/
40	private $pollingInterval;
41
42	/**
43	* @var Lock
44	*/
45	private $lock;
46
47	public function __construct(string $storeDir, int $preFetchCount, int $chmod, int $pollingInterval)
48	{
49	$fs = new Filesystem();
50	$fs->mkdir($storeDir);
51
52	$this->storeDir = $storeDir;
53	$this->preFetchCount = $preFetchCount;
54	$this->chmod = $chmod;
55	$this->pollingInterval = $pollingInterval;
56
57	$this->lock = new LegacyFilesystemLock();
58	}
59
60	/**
61	* @return FsMessage
62	*/
63	public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
64	{
65	return new FsMessage($body, $properties, $headers);
66	}
67
68	/**
69	* @return FsDestination
70	*/
71	public function createTopic(string $topicName): Topic
72	{
73	return $this->createQueue($topicName);
74	}
75
76	/**
77	* @return FsDestination
78	*/
79	public function createQueue(string $queueName): Queue
80	{
81	return new FsDestination(new \SplFileInfo($this->getStoreDir().'/'.$queueName));
82	}
83
84	public function declareDestination(FsDestination $destination): void
85	{
86	//InvalidDestinationException::assertDestinationInstanceOf($destination, FsDestination::class);
87
88	set_error_handler(function ($severity, $message, $file, $line) {
89	throw new \ErrorException($message, 0, $severity, $file, $line);
90	});
91
92	try {
93	if (false == file_exists((string) $destination->getFileInfo())) {
94	touch((string) $destination->getFileInfo());
95	chmod((string) $destination->getFileInfo(), $this->chmod);
96	}
97	} finally {
98	restore_error_handler();
99	}
100	}
101
102	public function workWithFile(FsDestination $destination, string $mode, callable $callback)
103	{
104	$this->declareDestination($destination);
105
106	set_error_handler(function ($severity, $message, $file, $line) {
107	throw new \ErrorException($message, 0, $severity, $file, $line);
108	}, E_ALL & ~E_USER_DEPRECATED);
109
110	try {
111	$file = fopen((string) $destination->getFileInfo(), $mode);
112	$this->lock->lock($destination);
113
114	return call_user_func($callback, $destination, $file);
115	} finally {
116	if (isset($file)) {
117	fclose($file);
118	}
119	$this->lock->release($destination);
120
121	restore_error_handler();
122	}
123	}
124
125	/**
126	* @return FsDestination
127	*/
128	public function createTemporaryQueue(): Queue
129	{
130	return new FsDestination(
131	new TempFile($this->getStoreDir().'/'.uniqid('tmp-q-', true))
132	);
133	}
134
135	/**
136	* @return FsProducer
137	*/
138	public function createProducer(): Producer
139	{
140	return new FsProducer($this);
141	}
142
143	/**
144	* @param FsDestination $destination
145	*
146	* @return FsConsumer
147	*/
148	public function createConsumer(Destination $destination): Consumer
149	{
150	InvalidDestinationException::assertDestinationInstanceOf($destination, FsDestination::class);
151
152	$consumer = new FsConsumer($this, $destination, $this->preFetchCount);
153
154	if ($this->pollingInterval) {
155	$consumer->setPollingInterval($this->pollingInterval);
156	}
157
158	return $consumer;
159	}
160
161	public function close(): void
162	{
163	$this->lock->releaseAll();
164	}
165
166	public function createSubscriptionConsumer(): SubscriptionConsumer
167	{
168	throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
169	}
170
171	/**
172	* @param FsDestination $queue
173	*/
174	public function purgeQueue(Queue $queue): void
175	{
176	InvalidDestinationException::assertDestinationInstanceOf($queue, FsDestination::class);
177
178	$this->workWithFile($queue, 'c', function (FsDestination $destination, $file) {
179	ftruncate($file, 0);
180	});
181	}
182
183	public function getPreFetchCount(): int
184	{
185	return $this->preFetchCount;
186	}
187
188	public function setPreFetchCount(int $preFetchCount): void
189	{
190	$this->preFetchCount = $preFetchCount;
191	}
192
193	private function getStoreDir(): string
194	{
195	if (false == is_dir($this->storeDir)) {
196	throw new \LogicException(sprintf('The directory %s does not exist', $this->storeDir));
197	}
198
199	if (false == is_writable($this->storeDir)) {
200	throw new \LogicException(sprintf('The directory %s is not writable', $this->storeDir));
201	}
202
203	return $this->storeDir;
204	}
205	}
206