implemented job retry function

jobs_retried.php

<?php
	// register Pheanstalk class loader
	require_once('pheanstalk/pheanstalk_init.php');

	$pheanstalk = new Pheanstalk('localhost');
	$pheanstalk
		->watch('touched')
		->ignore('default');

	$touched = array();

	while (true) {
		try {
			$job = $pheanstalk->reserve();

			if (!is_object($job)) {
				throw new Exception("Job is not defined.", 1);
				$pheanstalk
					->useTube('touched.problems')
					->put($job->getData());
				$pheanstalk
					->delete($job);
			}

			$data = json_decode($job->getData());

			if (!is_object($data)) {
				$pheanstalk
					->useTube('touched.problems')
					->put($job->getData());
				$pheanstalk
					->delete($job);
				throw new Exception("Job message is not an object.", 2);
			}

			$pid = $data->pid;
			$ref = $data->ref;

			if (!empty($pid)) {
				if(!touched($touched, $pid, $ref)) {
					$pheanstalk
						->useTube('touched.problems')
						->put($job->getData());
					$pheanstalk
						->delete($job);
					throw new Exception("Job is deleted after reaching 3 tries or timeout.", 3);
				}
				var_dump($touched);
				throw new Exception("Job pid does not match to 'somestupidpid'", 4);
			}

		} catch (Exception $e) {
			echo 	$e->getMessage() . ' in ' .
					$e->getFile() . ' at line ' .
					$e->getLine() . ' with code ' .
					$e->getCode() . PHP_EOL;
		}
		sleep(3);
	}

	function touched(&$touched, $pid, $ref) {
		if (isset($touched[$pid])) {
			if ($touched[$pid]['tries'] >= 3)
				return false;
			else if ((time() - $touched[$pid]['timestamp']) > (24 * 60 * 60))
				return false;
			else
				$touched[$pid]['tries'] = $touched[$pid]['tries'] + 1;
		}
		else
			$touched = array_merge($touched, array($pid => array('ref' => $ref, 'tries' => 1, 'timestamp' => time())));
		return true;
	}

?>