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

tenweb-speed-optimizer / vendor / enqueue / fs / FsConsumer.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

FsConsumer.php

215 lines

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Enqueue\Fs;
6
7	use Interop\Queue\Consumer;
8	use Interop\Queue\Exception\InvalidMessageException;
9	use Interop\Queue\Message;
10	use Interop\Queue\Queue;
11
12	class FsConsumer implements Consumer
13	{
14	/**
15	* @var FsDestination
16	*/
17	private $destination;
18
19	/**
20	* @var FsContext
21	*/
22	private $context;
23
24	/**
25	* @var int
26	*/
27	private $preFetchCount;
28
29	/**
30	* @var FsMessage[]
31	*/
32	private $preFetchedMessages;
33
34	/**
35	* In milliseconds.
36	*
37	* @var int
38	*/
39	private $pollingInterval = 100;
40
41	public function __construct(FsContext $context, FsDestination $destination, int $preFetchCount)
42	{
43	$this->context = $context;
44	$this->destination = $destination;
45	$this->preFetchCount = $preFetchCount;
46
47	$this->preFetchedMessages = [];
48	}
49
50	/**
51	* Set polling interval in milliseconds.
52	*/
53	public function setPollingInterval(int $msec): void
54	{
55	$this->pollingInterval = $msec;
56	}
57
58	/**
59	* Get polling interval in milliseconds.
60	*/
61	public function getPollingInterval(): int
62	{
63	return $this->pollingInterval;
64	}
65
66	/**
67	* @return FsDestination
68	*/
69	public function getQueue(): Queue
70	{
71	return $this->destination;
72	}
73
74	/**
75	* @return FsMessage
76	*/
77	public function receive(int $timeout = 0): ?Message
78	{
79	$timeout /= 1000;
80	$startAt = microtime(true);
81
82	while (true) {
83	$message = $this->receiveNoWait();
84
85	if ($message) {
86	return $message;
87	}
88
89	if ($timeout && (microtime(true) - $startAt) >= $timeout) {
90	return null;
91	}
92
93	usleep($this->pollingInterval * 1000);
94
95	if ($timeout && (microtime(true) - $startAt) >= $timeout) {
96	return null;
97	}
98	}
99	}
100
101	/**
102	* @return FsMessage
103	*/
104	public function receiveNoWait(): ?Message
105	{
106	if ($this->preFetchedMessages) {
107	return array_shift($this->preFetchedMessages);
108	}
109
110	$this->context->workWithFile($this->destination, 'c+', function (FsDestination $destination, $file) {
111	$count = $this->preFetchCount;
112	while ($count) {
113	$frame = $this->readFrame($file, 1);
114
115	//guards
116	if ($frame && false == ('\|' == $frame[0] \|\| ' ' == $frame[0])) {
117	throw new \LogicException(sprintf('The frame could start from either " " or "\|". The malformed frame starts with "%s".', $frame[0]));
118	}
119	if (0 !== $reminder = strlen($frame) % 64) {
120	throw new \LogicException(sprintf('The frame size is "%d" and it must divide exactly to 64 but it leaves a reminder "%d".', strlen($frame), $reminder));
121	}
122
123	ftruncate($file, fstat($file)['size'] - strlen($frame));
124	rewind($file);
125
126	$rawMessage = str_replace('\\|\{', '\|{', $frame);
127	$rawMessage = substr(trim($rawMessage), 1);
128
129	if ($rawMessage) {
130	try {
131	$fetchedMessage = FsMessage::jsonUnserialize($rawMessage);
132	$expireAt = $fetchedMessage->getHeader('x-expire-at');
133	if ($expireAt && $expireAt - microtime(true) < 0) {
134	// message has expired, just drop it.
135	return null;
136	}
137
138	$this->preFetchedMessages[] = $fetchedMessage;
139	} catch (\Exception $e) {
140	throw new \LogicException(sprintf("Cannot decode json message '%s'", $rawMessage), 0, $e);
141	}
142	} else {
143	return null;
144	}
145
146	--$count;
147	}
148	});
149
150	if ($this->preFetchedMessages) {
151	return array_shift($this->preFetchedMessages);
152	}
153
154	return null;
155	}
156
157	public function acknowledge(Message $message): void
158	{
159	// do nothing. fs transport always works in auto ack mode
160	}
161
162	public function reject(Message $message, bool $requeue = false): void
163	{
164	InvalidMessageException::assertMessageInstanceOf($message, FsMessage::class);
165
166	// do nothing on reject. fs transport always works in auto ack mode
167
168	if ($requeue) {
169	$this->context->createProducer()->send($this->destination, $message);
170	}
171	}
172
173	public function getPreFetchCount(): int
174	{
175	return $this->preFetchCount;
176	}
177
178	public function setPreFetchCount(int $preFetchCount): void
179	{
180	$this->preFetchCount = $preFetchCount;
181	}
182
183	/**
184	* @param resource $file
185	*/
186	private function readFrame($file, int $frameNumber): string
187	{
188	$frameSize = 64;
189	$offset = $frameNumber * $frameSize;
190
191	fseek($file, -$offset, SEEK_END);
192	$frame = fread($file, $frameSize);
193	if ('' == $frame) {
194	return '';
195	}
196
197	if (false !== strpos($frame, '\|{')) {
198	return $frame;
199	}
200
201	$previousFrame = $this->readFrame($file, $frameNumber + 1);
202
203	if ('\|' === substr($previousFrame, -1) && '{' === $frame[0]) {
204	$matched = [];
205	if (false === preg_match('/\ *?\\|$/', $previousFrame, $matched)) {
206	throw new \LogicException('Something went completely wrong.');
207	}
208
209	return $matched[0].$frame;
210	}
211
212	return $previousFrame.$frame;
213	}
214	}
215