DoctrineWorkerRunner::fetchNextCommandData() - Code Metrics - Inspection of "Refactored runner. More data in concurrency tests." - Gendoria/command-queue-doctrine-bundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0040e8...b35e00 )

by Tomasz

created 2016-11-03 14:16 UTC

DoctrineWorkerRunner::fetchNextCommandData() A

↳ Parent: DoctrineWorkerRunner

Complexity

Conditions	3
Paths	7

Size

Total Lines	22
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	3

Importance

Changes

Metric	Value
dl	0
loc	22
ccs	14
cts	14
cp	1
rs	9.2
c	0
b	0
f	0
cc	3
eloc	15
nc	7
nop	0
crap	3

<?php

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */

namespace Gendoria\CommandQueueDoctrineDriverBundle\Worker;

use DateTime;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\DBALException;
use Doctrine\DBAL\LockMode;
use Exception;
use Gendoria\CommandQueue\Worker\WorkerRunnerInterface;
use Gendoria\CommandQueueDoctrineDriverBundle\Worker\Exception\FetchException;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Symfony\Component\Console\Output\NullOutput;
use Symfony\Component\Console\Output\OutputInterface;

/**
 * Description of DoctrineWorkerRunner
 *
 * @author Tomasz Struczyński <[email protected]>
 */
class DoctrineWorkerRunner implements WorkerRunnerInterface
{

    /**
     * Number of maximum consecutive failed fetch operations before worker will fail permanently.
     * 
     * @var integer
     */
    const MAX_FAILED_FETCHES = 10;
    
    /**
     * Maximum process failures before command is discarded.
     * 
     * @var integer
     */
    const MAX_FAILED_PROCESS_ATTEMPTS = 10;
    
    /**
     * Doctrine worker.
     * 
     * @var DoctrineWorker
     */
    private $worker;

    /**
     * Database connection.
     * 
     * @var Connection
     */
    private $connection;

    /**
     * Table name.
     * 
     * @var string
     */
    private $tableName;

    /**
     * Pool.
     * 
     * @var string
     */
    private $pool;

    /**
     * Logger.
     * 
     * @var LoggerInterface
     */
    private $logger;
    
    /**
     * Number of consecutive failed fetches.
     * 
     * @var integer
     */
    private $failedFetches = 0;
    
    /**
     * Current options.
     * 
     * @var array
     */
    private $options = array();
    
    private $optionsProto = array(
        'run_times' => null,
        'exit_if_empty' => false,
        'sleep_intervals' => array(3000000, 6000000),
    );

    /**
     * Class constructor.
     * 
     * @param DoctrineWorker $worker
     * @param Connection $connection
     * @param string $tableName
     * @param string $pool
     */
    public function __construct(DoctrineWorker $worker, Connection $connection, $tableName, $pool, LoggerInterface $logger = null)
    {
        $this->worker = $worker;
        $this->connection = $connection;
        $this->tableName = $tableName;
        $this->pool = $pool;
        $this->logger = $logger ? $logger : new NullLogger();
    }

    public function run(array $options, OutputInterface $output = null)
    {
        if (!$output) {
            $output = new NullOutput();
        }
        $this->prepareOptions($options);
        $output->writeln(sprintf("Worker run with options: %s", print_r($this->options, true), OutputInterface::VERBOSITY_VERBOSE));
        $this->logger->debug(sprintf("Worker run with options: %s", print_r($this->options, true)));
        
        $this->connection->setTransactionIsolation(Connection::TRANSACTION_SERIALIZABLE);
        
        while ($this->checkRunTimes() && $this->runIteration($output, (bool)$this->options['exit_if_empty'])) {
            $this->logger->debug('Doctrine worker tick.');
        }
    }
    
    private function runIteration(OutputInterface $output, $exitIfEmpty = false)
    {
        try {
            $commandData = $this->fetchNext($output);
            if (is_array($commandData)) {
                $this->processNext($commandData, $output);
            } elseif ($exitIfEmpty) {
                return false;
            } else {
                $output->writeln("No messages to process yet.", OutputInterface::VERBOSITY_DEBUG);
                //Sleep to prevent hammering database
                $this->sleep();
            }
            return true;
        } catch (Exception $e) {
            return false;
        }
    }
    
    /**
     * Fetch next command data to process.
     * 
     * @param OutputInterface $output
     * @return array|boolean
     * @throws FetchException Thrown, when fetch failed permanently.
     */
    private function fetchNext(OutputInterface $output)
    {
        try {
            //Get one new row for update
            return $this->fetchNextCommandData();
        } catch (FetchException $e) {
            $this->logger->error('Exception while processing message.', array($e));
            $output->writeln(sprintf("<error>Error when processing message: %s</error>\n%s\n\n%s", $e->getMessage(), $e->getTraceAsString(), $e->getPrevious()->getTraceAsString()));
            if ($this->failedFetches > self::MAX_FAILED_FETCHES) {
                $output->writeln("<error>Too many consequent fetch errors - exiting.</error>");
                throw $e;
            }
            $this->sleep();
            return false;
        }
    }

    private function processNext($commandData, OutputInterface $output)
    {
        try {
            $output->writeln("Processing command.", OutputInterface::VERBOSITY_DEBUG);
            $this->worker->process($commandData);
            $this->connection->delete($this->tableName, array('id' => $commandData['id']));
        } catch (\Exception $e) {
            $this->logger->error('Exception while processing message.', array($e));
            $output->writeln(sprintf("<error>Error when processing message: %s</error>\n%s", $e->getMessage(), $e->getTraceAsString()));
            if ($commandData['failed_no'] >= self::MAX_FAILED_PROCESS_ATTEMPTS-1) {
                $this->logger->error('Command failed too many times, discarding.');
                $output->writeln(sprintf("<error>Command failed too many times, discarding.</error>"));
                $this->connection->delete($this->tableName, array('id' => $commandData['id']));
            } else {
                $this->setProcessedFailure($commandData['id'], $commandData['failed_no']+1);
            }
        }
    }

    /**
     * Fetch next command data.
     * 
     * @return array|boolean Array with command data, if there are pending commands; false otherwise.
     * @throws FetchException Thrown, when fetch resulted in an error (lock impossible).
     */
    private function fetchNextCommandData()
    {
        try {
            $this->connection->beginTransaction();
            $row = $this->connection->fetchAssoc($this->prepareFetchNextSql(), array(0, $this->pool, new DateTime()), array('integer', 'string', 'datetime'));

            if (empty($row)) {
                $this->connection->commit();
                return false;
            }

            $this->setProcessed($row['id']);
            
            $this->connection->commit();
            $this->failedFetches = 0;
            return $row;
        } catch (Exception $e) {
            $this->connection->rollBack();
            $this->failedFetches++;
            throw new FetchException("Exception while fetching data: ".$e->getMessage(), 500, $e);
        }
    }
    
    private function prepareFetchNextSql()
    {
        $platform = $this->connection->getDatabasePlatform();
        $sqlProto = "SELECT * "
            . " FROM " . $platform->appendLockHint($this->tableName, LockMode::PESSIMISTIC_WRITE)
            . " WHERE processed = ? AND pool = ? AND process_after <= ? "
            . " ORDER BY id ASC"
        ;
        return $platform->modifyLimitQuery($sqlProto, 1) . " " . $platform->getWriteLockSQL();
    }
    
    /**
     * Sleep to prevent hammering database.
     * 
     * Sleep interval is somewhat random to minify risk of database collisions.
     * 
     * @return void
     */
    private function sleep()
    {
        usleep(mt_rand($this->options['sleep_intervals'][0], $this->options['sleep_intervals'][1]));
    }
    
    /**
     * Set processed status for row.
     * 
     * @param integer $id
     * @return integer Number of affected rows.

     * @throws DBALException Thrown, if race condition has been detected or other database error occurred.
     */
    private function setProcessed($id)
    {
        $parameters = array(
            'processed' => true,
            'id' => (int)$id,
        );
        $types = array(
            'processed' => 'boolean',
            'id' => 'smallint'
        );
        $updateSql = "UPDATE " . $this->tableName
            . " SET processed = :processed"
            . " WHERE id = :id";
        if ($this->connection->executeUpdate($updateSql, $parameters, $types) !== 1) {
            throw new DBALException("Race condition detected. Aborting.");
        }
    }
    
    /**
     * Update processed status for row.
     * 
     * @param integer $id
     * @param integer $failedRetries
     * @return integer
     */
    private function setProcessedFailure($id, $failedRetries)
    {
        $parameters = array(
            'processed' => false,
            'failed_no' => (int)$failedRetries,
            'id' => (int)$id,
            'process_after' => new DateTime('@'.(time()+20*$failedRetries)),
        );
        $types = array(
            'processed' => 'boolean',
            'failed_no' => 'integer',
            'id' => 'smallint',
            'process_after' => 'datetime',
        );
        $updateSql = "UPDATE " . $this->tableName
            . " SET processed = :processed, failed_no = :failed_no, process_after = :process_after"
            . " WHERE id = :id";
        return $this->connection->executeUpdate($updateSql, $parameters, $types);
    }
    
    private function prepareOptions($options) {
        $this->options = array_merge($this->optionsProto, $options);
    }
    
    private function checkRunTimes()
    {
        return !($this->options['run_times'] !== null && $this->options['run_times']-- == 0);
    }
}


1		<?php
2
3		/*
4		* To change this license header, choose License Headers in Project Properties.
5		* To change this template file, choose Tools \| Templates
6		* and open the template in the editor.
7		*/
8
9		namespace Gendoria\CommandQueueDoctrineDriverBundle\Worker;
10
11		use DateTime;
12		use Doctrine\DBAL\Connection;
13		use Doctrine\DBAL\DBALException;
14		use Doctrine\DBAL\LockMode;
15		use Exception;
16		use Gendoria\CommandQueue\Worker\WorkerRunnerInterface;
17		use Gendoria\CommandQueueDoctrineDriverBundle\Worker\Exception\FetchException;
18		use Psr\Log\LoggerInterface;
19		use Psr\Log\NullLogger;
20		use Symfony\Component\Console\Output\NullOutput;
21		use Symfony\Component\Console\Output\OutputInterface;
22
23		/**
24		* Description of DoctrineWorkerRunner
25		*
26		* @author Tomasz Struczyński <[email protected]>
27		*/
28		class DoctrineWorkerRunner implements WorkerRunnerInterface
29		{
30
31		/**
32		* Number of maximum consecutive failed fetch operations before worker will fail permanently.
33		*
34		* @var integer
35		*/
36		const MAX_FAILED_FETCHES = 10;
37
38		/**
39		* Maximum process failures before command is discarded.
40		*
41		* @var integer
42		*/
43		const MAX_FAILED_PROCESS_ATTEMPTS = 10;
44
45		/**
46		* Doctrine worker.
47		*
48		* @var DoctrineWorker
49		*/
50		private $worker;
51
52		/**
53		* Database connection.
54		*
55		* @var Connection
56		*/
57		private $connection;
58
59		/**
60		* Table name.
61		*
62		* @var string
63		*/
64		private $tableName;
65
66		/**
67		* Pool.
68		*
69		* @var string
70		*/
71		private $pool;
72
73		/**
74		* Logger.
75		*
76		* @var LoggerInterface
77		*/
78		private $logger;
79
80		/**
81		* Number of consecutive failed fetches.
82		*
83		* @var integer
84		*/
85		private $failedFetches = 0;
86
87		/**
88		* Current options.
89		*
90		* @var array
91		*/
92		private $options = array();
93
94		private $optionsProto = array(
95		'run_times' => null,
96		'exit_if_empty' => false,
97		'sleep_intervals' => array(3000000, 6000000),
98		);
99
100		/**
101		* Class constructor.
102		*
103		* @param DoctrineWorker $worker
104		* @param Connection $connection
105		* @param string $tableName
106		* @param string $pool
107		*/
108	5	public function __construct(DoctrineWorker $worker, Connection $connection, $tableName, $pool, LoggerInterface $logger = null)
109		{
110	5	$this->worker = $worker;
111	5	$this->connection = $connection;
112	5	$this->tableName = $tableName;
113	5	$this->pool = $pool;
114	5	$this->logger = $logger ? $logger : new NullLogger();
115	5	}
116
117	5	public function run(array $options, OutputInterface $output = null)
118		{
119	5	if (!$output) {
120	1	$output = new NullOutput();
121	1	}
122	5	$this->prepareOptions($options);
123	5	$output->writeln(sprintf("Worker run with options: %s", print_r($this->options, true), OutputInterface::VERBOSITY_VERBOSE));
124	5	$this->logger->debug(sprintf("Worker run with options: %s", print_r($this->options, true)));
125
126	5	$this->connection->setTransactionIsolation(Connection::TRANSACTION_SERIALIZABLE);
127
128	5	while ($this->checkRunTimes() && $this->runIteration($output, (bool)$this->options['exit_if_empty'])) {
129	4	$this->logger->debug('Doctrine worker tick.');
130	4	}
131	5	}
132
133	5	private function runIteration(OutputInterface $output, $exitIfEmpty = false)
134		{
135		try {
136	5	$commandData = $this->fetchNext($output);
137	5	if (is_array($commandData)) {
138	3	$this->processNext($commandData, $output);
139	5	} elseif ($exitIfEmpty) {
140	1	return false;
141		} else {
142	3	$output->writeln("No messages to process yet.", OutputInterface::VERBOSITY_DEBUG);
143		//Sleep to prevent hammering database
144	3	$this->sleep();
145		}
146	4	return true;
147	1	} catch (Exception $e) {
148	1	return false;
149		}
150		}
151
152		/**
153		* Fetch next command data to process.
154		*
155		* @param OutputInterface $output
156		* @return array\|boolean
157		* @throws FetchException Thrown, when fetch failed permanently.
158		*/
159	5	private function fetchNext(OutputInterface $output)
160		{
161		try {
162		//Get one new row for update
163	5	return $this->fetchNextCommandData();
164	1	} catch (FetchException $e) {
165	1	$this->logger->error('Exception while processing message.', array($e));
166	1	$output->writeln(sprintf("<error>Error when processing message: %s</error>\n%s\n\n%s", $e->getMessage(), $e->getTraceAsString(), $e->getPrevious()->getTraceAsString()));
167	1	if ($this->failedFetches > self::MAX_FAILED_FETCHES) {
168	1	$output->writeln("<error>Too many consequent fetch errors - exiting.</error>");
169	1	throw $e;
170		}
171	1	$this->sleep();
172	1	return false;
173		}
174		}
175
176	3	private function processNext($commandData, OutputInterface $output)
177		{
178		try {
179	3	$output->writeln("Processing command.", OutputInterface::VERBOSITY_DEBUG);
180	3	$this->worker->process($commandData);
181	2	$this->connection->delete($this->tableName, array('id' => $commandData['id']));
182	3	} catch (\Exception $e) {
183	1	$this->logger->error('Exception while processing message.', array($e));
184	1	$output->writeln(sprintf("<error>Error when processing message: %s</error>\n%s", $e->getMessage(), $e->getTraceAsString()));
185	1	if ($commandData['failed_no'] >= self::MAX_FAILED_PROCESS_ATTEMPTS-1) {
186	1	$this->logger->error('Command failed too many times, discarding.');
187	1	$output->writeln(sprintf("<error>Command failed too many times, discarding.</error>"));
188	1	$this->connection->delete($this->tableName, array('id' => $commandData['id']));
189	1	} else {
190	1	$this->setProcessedFailure($commandData['id'], $commandData['failed_no']+1);
191		}
192		}
193	3	}
194
195		/**
196		* Fetch next command data.
197		*
198		* @return array\|boolean Array with command data, if there are pending commands; false otherwise.
199		* @throws FetchException Thrown, when fetch resulted in an error (lock impossible).
200		*/
201	5	private function fetchNextCommandData()
202		{
203		try {
204	5	$this->connection->beginTransaction();
205	5	$row = $this->connection->fetchAssoc($this->prepareFetchNextSql(), array(0, $this->pool, new DateTime()), array('integer', 'string', 'datetime'));
206
207	5	if (empty($row)) {
208	3	$this->connection->commit();
209	3	return false;
210		}
211
212	4	$this->setProcessed($row['id']);
213
214	3	$this->connection->commit();
215	3	$this->failedFetches = 0;
216	3	return $row;
217	1	} catch (Exception $e) {
218	1	$this->connection->rollBack();
219	1	$this->failedFetches++;
220	1	throw new FetchException("Exception while fetching data: ".$e->getMessage(), 500, $e);
221		}
222		}
223
224	5	private function prepareFetchNextSql()
225		{
226	5	$platform = $this->connection->getDatabasePlatform();
227		$sqlProto = "SELECT * "
228	5	. " FROM " . $platform->appendLockHint($this->tableName, LockMode::PESSIMISTIC_WRITE)
229	5	. " WHERE processed = ? AND pool = ? AND process_after <= ? "
230	5	. " ORDER BY id ASC"
231	5	;
232	5	return $platform->modifyLimitQuery($sqlProto, 1) . " " . $platform->getWriteLockSQL();
233		}
234
235		/**
236		* Sleep to prevent hammering database.
237		*
238		* Sleep interval is somewhat random to minify risk of database collisions.
239		*
240		* @return void
241		*/
242	3	private function sleep()
243		{
244	3	usleep(mt_rand($this->options['sleep_intervals'][0], $this->options['sleep_intervals'][1]));
245	3	}
246
247		/**
248		* Set processed status for row.
249		*
250		* @param integer $id
251		* @return integer Number of affected rows.
		0 ignored issues – show Documentation introduced 2016-11-03 14:21 UTC by Report Bug Copy Issue Report Should the return type not be `integer\|null`? This check compares the return type specified in the `@return` annotation of a function or method doc comment with the types returned by the function and raises an issue if they mismatch. Loading history...
252		* @throws DBALException Thrown, if race condition has been detected or other database error occurred.
253		*/
254	4	private function setProcessed($id)
255		{
256		$parameters = array(
257	4	'processed' => true,
258	4	'id' => (int)$id,
259	4	);
260		$types = array(
261	4	'processed' => 'boolean',
262		'id' => 'smallint'
263	4	);
264	4	$updateSql = "UPDATE " . $this->tableName
265	4	. " SET processed = :processed"
266	4	. " WHERE id = :id";
267	4	if ($this->connection->executeUpdate($updateSql, $parameters, $types) !== 1) {
268	1	throw new DBALException("Race condition detected. Aborting.");
269		}
270	3	}
271
272		/**
273		* Update processed status for row.
274		*
275		* @param integer $id
276		* @param integer $failedRetries
277		* @return integer
278		*/
279	1	private function setProcessedFailure($id, $failedRetries)
280		{
281		$parameters = array(
282	1	'processed' => false,
283	1	'failed_no' => (int)$failedRetries,
284	1	'id' => (int)$id,
285	1	'process_after' => new DateTime('@'.(time()+20*$failedRetries)),
286	1	);
287		$types = array(
288	1	'processed' => 'boolean',
289	1	'failed_no' => 'integer',
290	1	'id' => 'smallint',
291	1	'process_after' => 'datetime',
292	1	);
293	1	$updateSql = "UPDATE " . $this->tableName
294	1	. " SET processed = :processed, failed_no = :failed_no, process_after = :process_after"
295	1	. " WHERE id = :id";
296	1	return $this->connection->executeUpdate($updateSql, $parameters, $types);
297		}
298
299	5	private function prepareOptions($options) {
300	5	$this->options = array_merge($this->optionsProto, $options);
301	5	}
302
303	5	private function checkRunTimes()
304		{
305	5	return !($this->options['run_times'] !== null && $this->options['run_times']-- == 0);
306		}
307		}
308

Gendoria / command-queue-doctrine-bundle

Push — master ( 0040e8...b35e00 )

DoctrineWorkerRunner::fetchNextCommandData() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like