Skip to content

Commit b911c79

Browse files
author
Bastian Waidelich
authored
Merge pull request #4 from bwaidelich/master
FEATURE: Adjust to major overhaul of Flowpack.JobQueue.Common package
2 parents 66e35b0 + 9c2f4db commit b911c79

File tree

5 files changed

+179
-240
lines changed

5 files changed

+179
-240
lines changed

‎Classes/Queue/RedisQueue.php‎

Lines changed: 92 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313

1414
use Flowpack\JobQueue\Common\Queue\Message;
1515
use Flowpack\JobQueue\Common\Queue\QueueInterface;
16+
use TYPO3\Flow\Utility\Algorithms;
17+
use Flowpack\JobQueue\Common\Exception as JobQueueException;
1618

1719
/**
1820
* A queue implementation using Redis as the queue backend
@@ -58,51 +60,49 @@ class RedisQueue implements QueueInterface
5860
protected $maxReconnectDelay = 30.0;
5961

6062
/**
61-
* Constructor
62-
*
6363
* @param string $name
6464
* @param array $options
65+
* @throws JobQueueException
6566
*/
66-
public function __construct($name, array $options = array())
67+
public function __construct($name, array $options = [])
6768
{
6869
$this->name = $name;
6970
if (isset($options['defaultTimeout'])) {
7071
$this->defaultTimeout = (integer)$options['defaultTimeout'];
7172
}
72-
$this->clientOptions = isset($options['client']) ? $options['client'] : array();
73-
73+
$this->clientOptions = isset($options['client']) ? $options['client'] : [];
7474
$this->client = new \Redis();
7575
if (!$this->connectClient()) {
76-
throw new \Flowpack\JobQueue\Common\Exception('Could not connect to Redis', 1467382685);
76+
throw new JobQueueException('Could not connect to Redis', 1467382685);
7777
}
7878
}
7979

8080
/**
81-
* Submit a message to the queue
82-
*
83-
* @param Message $message
84-
* @return void
81+
* @inheritdoc
82+
*/
83+
public function getName()
84+
{
85+
return $this->name;
86+
}
87+
88+
/**
89+
* @inheritdoc
8590
*/
86-
public function submit(Message $message)
91+
public function submit($payload, array $options = [])
8792
{
8893
$this->checkClientConnection();
89-
if ($message->getIdentifier() !== null) {
90-
$added = $this->client->sAdd("queue:{$this->name}:ids", $message->getIdentifier());
91-
if (!$added) {
92-
return;
93-
}
94+
$messageId = Algorithms::generateUUID();
95+
$idStored = $this->client->hSet("queue:{$this->name}:ids", $messageId, json_encode($payload));
96+
if ($idStored === 0) {
97+
return null;
9498
}
95-
$encodedMessage = $this->encodeMessage($message);
96-
$this->client->lPush("queue:{$this->name}:messages", $encodedMessage);
97-
$message->setState(Message::STATE_SUBMITTED);
99+
100+
$this->client->lPush("queue:{$this->name}:messages", $messageId);
101+
return $messageId;
98102
}
99103

100104
/**
101-
* Wait for a message in the queue and return the message for processing
102-
* (without safety queue)
103-
*
104-
* @param int $timeout
105-
* @return Message The received message or NULL if a timeout occurred
105+
* @inheritdoc
106106
*/
107107
public function waitAndTake($timeout = null)
108108
{
@@ -111,149 +111,128 @@ public function waitAndTake($timeout = null)
111111
}
112112
$this->checkClientConnection();
113113
$keyAndValue = $this->client->brPop("queue:{$this->name}:messages", $timeout);
114-
$value = isset($keyAndValue[1]) ? $keyAndValue[1] : null;
115-
if (is_string($value)) {
116-
$message = $this->decodeMessage($value);
117-
118-
if ($message->getIdentifier() !== null) {
119-
$this->client->sRem("queue:{$this->name}:ids", $message->getIdentifier());
120-
}
121-
122-
// The message is marked as done
123-
$message->setState(Message::STATE_DONE);
124-
125-
return $message;
126-
} else {
114+
$messageId = isset($keyAndValue[1]) ? $keyAndValue[1] : null;
115+
if ($messageId === null) {
127116
return null;
128117
}
118+
$message = $this->getMessageById($messageId);
119+
if ($message !== null) {
120+
$this->client->hDel("queue:{$this->name}:ids", $messageId);
121+
}
122+
return $message;
129123
}
130124

131125
/**
132-
* Wait for a message in the queue and save the message to a safety queue
133-
*
134-
* TODO: Idea for implementing a TTR (time to run) with monitoring of safety queue. E.g.
135-
* use different queue names with encoded times? With "brpoplpush" we cannot modify the
136-
* queued item on transfer to the safety queue and we cannot update a timestamp to mark
137-
* the run start time in the message, so separate keys should be used for this.
138-
*
139-
* @param int $timeout
140-
* @return Message
126+
* @inheritdoc
141127
*/
142128
public function waitAndReserve($timeout = null)
143129
{
144130
if ($timeout === null) {
145131
$timeout = $this->defaultTimeout;
146132
}
147133
$this->checkClientConnection();
148-
$value = $this->client->brpoplpush("queue:{$this->name}:messages", "queue:{$this->name}:processing", $timeout);
149-
if (is_string($value)) {
150-
$message = $this->decodeMessage($value);
151-
if ($message->getIdentifier() !== null) {
152-
$this->client->sRem("queue:{$this->name}:ids", $message->getIdentifier());
153-
}
154-
return $message;
155-
} else {
156-
return null;
157-
}
134+
$messageId = $this->client->brpoplpush("queue:{$this->name}:messages", "queue:{$this->name}:processing", $timeout);
135+
return $this->getMessageById($messageId);
158136
}
159137

160138
/**
161-
* Mark a message as finished
162-
*
163-
* @param Message $message
164-
* @return boolean TRUE if the message could be removed
139+
* @inheritdoc
140+
*/
141+
public function release($messageId, array $options = [])
142+
{
143+
$this->checkClientConnection();
144+
$this->client->lRem("queue:{$this->name}:processing", $messageId, 0);
145+
$numberOfReleases = (integer)$this->client->hGet("queue:{$this->name}:releases", $messageId);
146+
$this->client->hSet("queue:{$this->name}:releases", $messageId, $numberOfReleases + 1);
147+
$this->client->lPush("queue:{$this->name}:messages", $messageId);
148+
}
149+
150+
/**
151+
* @inheritdoc
165152
*/
166-
public function finish(Message $message)
153+
public function abort($messageId)
167154
{
168155
$this->checkClientConnection();
169-
$originalValue = $message->getOriginalValue();
170-
$success = $this->client->lRem("queue:{$this->name}:processing", $originalValue, 0) > 0;
171-
if ($success) {
172-
$message->setState(Message::STATE_DONE);
156+
$numberOfRemoved = $this->client->lRem("queue:{$this->name}:processing", $messageId, 0);
157+
if ($numberOfRemoved === 1) {
158+
$this->client->lPush("queue:{$this->name}:failed", $messageId);
173159
}
174-
return $success;
175160
}
176161

177162
/**
178-
* Peek for messages
179-
*
180-
* @param integer $limit
181-
* @return Message[] Messages or empty array if no messages were present
163+
* @inheritdoc
164+
*/
165+
public function finish($messageId)
166+
{
167+
$this->checkClientConnection();
168+
$this->client->hDel("queue:{$this->name}:ids", $messageId);
169+
$this->client->hDel("queue:{$this->name}:releases", $messageId);
170+
return $this->client->lRem("queue:{$this->name}:processing", $messageId, 0) > 0;
171+
}
172+
173+
/**
174+
* @inheritdoc
182175
*/
183176
public function peek($limit = 1)
184177
{
185178
$this->checkClientConnection();
186179
$result = $this->client->lRange("queue:{$this->name}:messages", -($limit), -1);
187-
if (is_array($result) && count($result) > 0) {
188-
$messages = array();
189-
foreach ($result as $value) {
190-
$message = $this->decodeMessage($value);
191-
// The message is still submitted and should not be processed!
192-
$message->setState(Message::STATE_SUBMITTED);
193-
$messages[] = $message;
194-
}
195-
return $messages;
180+
if (!is_array($result) || count($result) === 0) {
181+
return [];
182+
}
183+
$messages = [];
184+
foreach ($result as $messageId) {
185+
$encodedPayload = $this->client->hGet("queue:{$this->name}:ids", $messageId);
186+
$messages[] = new Message($messageId, json_decode($encodedPayload, true));
196187
}
197-
return array();
188+
return $messages;
198189
}
199190

200191
/**
201-
* Count messages in the queue
202-
*
203-
* @return integer
192+
* @inheritdoc
204193
*/
205194
public function count()
206195
{
207196
$this->checkClientConnection();
208-
$count = $this->client->lLen("queue:{$this->name}:messages");
209-
return $count;
197+
return $this->client->lLen("queue:{$this->name}:messages");
210198
}
211199

212200
/**
213-
* Encode a message
214-
*
215-
* Updates the original value property of the message to resemble the
216-
* encoded representation.
217-
*
218-
* @param Message $message
219-
* @return string
201+
* @return void
220202
*/
221-
protected function encodeMessage(Message $message)
203+
public function setUp()
222204
{
223-
$value = json_encode($message->toArray());
224-
$message->setOriginalValue($value);
225-
return $value;
205+
$this->checkClientConnection();
226206
}
227207

228208
/**
229-
* Decode a message from a string representation
230-
*
231-
* @param string $value
232-
* @return Message
209+
* @inheritdoc
233210
*/
234-
protected function decodeMessage($value)
211+
public function flush()
235212
{
236-
$decodedMessage = json_decode($value, true);
237-
$message = new Message($decodedMessage['payload']);
238-
if (isset($decodedMessage['identifier'])) {
239-
$message->setIdentifier($decodedMessage['identifier']);
240-
}
241-
$message->setOriginalValue($value);
242-
return $message;
213+
$this->checkClientConnection();
214+
$this->client->flushDB();
243215
}
244216

245217
/**
246-
*
247-
* @param string $identifier
218+
* @param string $messageId
248219
* @return Message
249220
*/
250-
public function getMessage($identifier)
221+
protected function getMessageById($messageId)
251222
{
252-
return null;
223+
if (!is_string($messageId)) {
224+
return null;
225+
}
226+
$encodedPayload = $this->client->hGet("queue:{$this->name}:ids", $messageId);
227+
$numberOfReleases = (integer)$this->client->hGet("queue:{$this->name}:releases", $messageId);
228+
return new Message($messageId, json_decode($encodedPayload, true), $numberOfReleases);
253229
}
254230

255231
/**
256232
* Check if the Redis client connection is still up and reconnect if Redis was disconnected
233+
*
234+
* @return void
235+
* @throws JobQueueException
257236
*/
258237
protected function checkClientConnection()
259238
{
@@ -268,7 +247,7 @@ protected function checkClientConnection()
268247
}
269248
if ($reconnect) {
270249
if (!$this->connectClient()) {
271-
throw new \Flowpack\JobQueue\Common\Exception('Could not connect to Redis', 1467382685);
250+
throw new JobQueueException('Could not connect to Redis', 1467382685);
272251
}
273252
}
274253
}
@@ -286,20 +265,16 @@ protected function connectClient()
286265
$host = isset($this->clientOptions['host']) ? $this->clientOptions['host'] : '127.0.0.1';
287266
$port = isset($this->clientOptions['port']) ? $this->clientOptions['port'] : 6379;
288267
$database = isset($this->clientOptions['database']) ? $this->clientOptions['database'] : 0;
289-
290268
// The connection read timeout should be higher than the timeout for blocking operations!
291269
$timeout = isset($this->clientOptions['timeout']) ? $this->clientOptions['timeout'] : round($this->defaultTimeout * 1.5);
292270
$connected = $this->client->connect($host, $port, $timeout) && $this->client->select($database);
293-
294271
// Break the cycle that could cause a high CPU load
295272
if (!$connected) {
296273
usleep($this->reconnectDelay * 1e6);
297274
$this->reconnectDelay = min($this->reconnectDelay * $this->reconnectDecay, $this->maxReconnectDelay);
298275
} else {
299276
$this->reconnectDelay = 1.0;
300277
}
301-
302278
return $connected;
303279
}
304-
305280
}

‎CodeOfConduct.rst‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
Contributor Code of Conduct
2+
---------------------------
3+
4+
As contributors and maintainers of this project, and in the interest of fostering an open and welcoming community, we pledge to respect all people who contribute through reporting issues, posting feature requests, updating documentation, submitting pull requests or patches, and other activities.
5+
6+
We are committed to making participation in this project a harassment-free experience for everyone, regardless of level of experience, gender, gender identity and expression, sexual orientation, disability, personal appearance, body size, race, ethnicity, age, religion, or nationality.
7+
8+
Examples of unacceptable behavior by participants include:
9+
10+
* The use of sexualized language or imagery
11+
* Personal attacks
12+
* Trolling or insulting/derogatory comments
13+
* Public or private harassment
14+
* Publishing other's private information, such as physical or electronic addresses, without explicit permission
15+
* Other unethical or unprofessional conduct.
16+
17+
Project maintainers have the right and responsibility to remove, edit, or reject comments, commits, code, wiki edits, issues, and other contributions that are not aligned to this Code of Conduct. By adopting this Code of Conduct, project maintainers commit themselves to fairly and consistently applying these principles to every aspect of managing this project. Project maintainers who do not follow or enforce the Code of Conduct may be permanently removed from the project team.
18+
19+
This code of conduct applies both within project spaces and in public spaces when an individual is representing the project or its community.
20+
21+
Instances of abusive, harassing, or otherwise unacceptable behavior may be reported by opening an issue or contacting one or more of the project maintainers.
22+
23+
This Code of Conduct is adapted from the `Contributor Covenant <http://contributor-covenant.org>`_, version 1.2.0, available at (http://contributor-covenant.org/version/1/2/0/

‎LICENSE‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
The MIT License (MIT)
22

3-
Copyright (c) 2015 Neos project contributors
3+
Copyright (c) 2016 Neos project contributors
44

55
Permission is hereby granted, free of charge, to any person obtaining a copy
66
of this software and associated documentation files (the "Software"), to deal

0 commit comments

Comments
 (0)