GearmanExecute.php 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321
  1. <?php
  2. /**
  3. * Gearman Bundle for Symfony2
  4. *
  5. * For the full copyright and license information, please view the LICENSE
  6. * file that was distributed with this source code.
  7. *
  8. * Feel free to edit as you please, and have fun.
  9. *
  10. * @author Marc Morera <yuhu@mmoreram.com>
  11. */
  12. namespace Mmoreram\GearmanBundle\Service;
  13. use Symfony\Component\Console\Output\NullOutput;
  14. use Symfony\Component\Console\Output\OutputInterface;
  15. use Symfony\Component\DependencyInjection\ContainerAwareInterface;
  16. use Symfony\Component\DependencyInjection\ContainerInterface;
  17. use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  18. use Mmoreram\GearmanBundle\Command\Util\GearmanOutputAwareInterface;
  19. use Mmoreram\GearmanBundle\Event\GearmanWorkExecutedEvent;
  20. use Mmoreram\GearmanBundle\GearmanEvents;
  21. use Mmoreram\GearmanBundle\Service\Abstracts\AbstractGearmanService;
  22. /**
  23. * Gearman execute methods. All Worker methods
  24. *
  25. * @since 2.3.1
  26. */
  27. class GearmanExecute extends AbstractGearmanService
  28. {
  29. /**
  30. * @var ContainerInterface
  31. *
  32. * Container instance
  33. */
  34. private $container;
  35. /**
  36. * @var EventDispatcherInterface
  37. *
  38. * EventDispatcher instance
  39. */
  40. protected $eventDispatcher;
  41. /**
  42. * @var OutputInterface
  43. *
  44. * Output instance
  45. */
  46. protected $output;
  47. /**
  48. * Set container
  49. *
  50. * @param ContainerInterface $container Container
  51. *
  52. * @return GearmanExecute self Object
  53. */
  54. public function setContainer(ContainerInterface $container)
  55. {
  56. $this->container = $container;
  57. return $this;
  58. }
  59. /**
  60. * Set event dispatcher
  61. *
  62. * @param EventDispatcherInterface $eventDispatcher
  63. *
  64. * @return GearmanExecute self Object
  65. */
  66. public function setEventDispatcher(EventDispatcherInterface $eventDispatcher)
  67. {
  68. $this->eventDispatcher = $eventDispatcher;
  69. return $this;
  70. }
  71. /**
  72. * Set output
  73. *
  74. * @param OutputInterface $output
  75. *
  76. * @return GearmanExecute self Object
  77. */
  78. public function setOutput(OutputInterface $output)
  79. {
  80. $this->output = $output;
  81. return $this;
  82. }
  83. /**
  84. * Executes a job given a jobName and given settings and annotations of job
  85. *
  86. * @param string $jobName Name of job to be executed
  87. * @param GearmanWorker $gearmanWorker Worker instance to use
  88. */
  89. public function executeJob($jobName, \GearmanWorker $gearmanWorker = null)
  90. {
  91. $worker = $this->getJob($jobName);
  92. if (false !== $worker) {
  93. $this->callJob($worker, $gearmanWorker);
  94. }
  95. }
  96. /**
  97. * Given a worker, execute GearmanWorker function defined by job.
  98. *
  99. * @param array $worker Worker definition
  100. * @param GearmanWorker $gearmanWorker Worker instance to use
  101. * @return GearmanExecute self Object
  102. */
  103. private function callJob(Array $worker, \GearmanWorker $gearmanWorker = null)
  104. {
  105. if(is_null($gearmanWorker)){
  106. $gearmanWorker = new \GearmanWorker;
  107. }
  108. if (isset($worker['job'])) {
  109. $jobs = array($worker['job']);
  110. $iterations = $worker['job']['iterations'];
  111. $this->addServers($gearmanWorker, $worker['job']['servers']);
  112. } else {
  113. $jobs = $worker['jobs'];
  114. $iterations = $worker['iterations'];
  115. $this->addServers($gearmanWorker, $worker['servers']);
  116. }
  117. $objInstance = $this->createJob($worker);
  118. $this->runJob($gearmanWorker, $objInstance, $jobs, $iterations);
  119. return $this;
  120. }
  121. /**
  122. * Given a worker settings, return Job instance
  123. *
  124. * @param array $worker Worker settings
  125. *
  126. * @return Object Job instance
  127. */
  128. private function createJob(array $worker)
  129. {
  130. /**
  131. * If service is defined, we must retrieve this class with dependency injection
  132. *
  133. * Otherwise we just create it with a simple new()
  134. */
  135. if ($worker['service']) {
  136. $objInstance = $this->container->get($worker['service']);
  137. } else {
  138. $objInstance = new $worker['className'];
  139. /**
  140. * If instance of given object is instanceof
  141. * ContainerAwareInterface, we inject full container by calling
  142. * container setter.
  143. *
  144. * @see https://github.com/mmoreram/gearman-bundle/pull/12
  145. */
  146. if ($objInstance instanceof ContainerAwareInterface) {
  147. $objInstance->setContainer($this->container);
  148. }
  149. }
  150. return $objInstance;
  151. }
  152. /**
  153. * Given a GearmanWorker and an instance of Job, run it
  154. *
  155. * @param \GearmanWorker $gearmanWorker Gearman Worker
  156. * @param Object $objInstance Job instance
  157. * @param array $jobs Array of jobs to subscribe
  158. * @param integer $iterations Number of iterations
  159. *
  160. * @return GearmanExecute self Object
  161. */
  162. private function runJob(\GearmanWorker $gearmanWorker, $objInstance, array $jobs, $iterations)
  163. {
  164. /**
  165. * Set the output of this instance, this should allow workers to use the console output.
  166. */
  167. if ($objInstance instanceof GearmanOutputAwareInterface) {
  168. $objInstance->setOutput($this->output ? : new NullOutput());
  169. }
  170. /**
  171. * Every job defined in worker is added into GearmanWorker
  172. */
  173. foreach ($jobs as $job) {
  174. $gearmanWorker->addFunction(
  175. $job['realCallableName'],
  176. array($this, 'handleJob'),
  177. array(
  178. 'job_object_instance' => $objInstance,
  179. 'job_method' => $job['methodName'],
  180. )
  181. );
  182. }
  183. /**
  184. * If iterations value is 0, is like worker will never die
  185. */
  186. $alive = (0 == $iterations);
  187. /**
  188. * Executes GearmanWorker with all jobs defined
  189. */
  190. while ($gearmanWorker->work()) {
  191. $iterations--;
  192. $event = new GearmanWorkExecutedEvent($jobs, $iterations, $gearmanWorker->returnCode());
  193. $this->eventDispatcher->dispatch(GearmanEvents::GEARMAN_WORK_EXECUTED, $event);
  194. if ($gearmanWorker->returnCode() != GEARMAN_SUCCESS) {
  195. break;
  196. }
  197. /**
  198. * Only finishes its execution if alive is false and iterations
  199. * arrives to 0
  200. */
  201. if (!$alive && $iterations <= 0) {
  202. break;
  203. }
  204. }
  205. }
  206. /**
  207. * Adds into worker all defined Servers.
  208. * If any is defined, performs default method
  209. *
  210. * @param \GearmanWorker $gmworker Worker to perform configuration
  211. * @param array $servers Servers array
  212. */
  213. private function addServers(\GearmanWorker $gmworker, Array $servers)
  214. {
  215. if (!empty($servers)) {
  216. foreach ($servers as $server) {
  217. $gmworker->addServer($server['host'], $server['port']);
  218. }
  219. } else {
  220. $gmworker->addServer();
  221. }
  222. }
  223. /**
  224. * Executes a worker given a workerName subscribing all his jobs inside and
  225. * given settings and annotations of worker and jobs
  226. *
  227. * @param string $workerName Name of worker to be executed
  228. */
  229. public function executeWorker($workerName)
  230. {
  231. $worker = $this->getWorker($workerName);
  232. if (false !== $worker) {
  233. $this->callJob($worker);
  234. }
  235. }
  236. /**
  237. * Wrapper function handler for all registered functions
  238. * This allows us to do some nice logging when jobs are started/finished
  239. *
  240. * @see https://github.com/brianlmoon/GearmanManager/blob/ffc828dac2547aff76cb4962bb3fcc4f454ec8a2/GearmanPeclManager.php#L95-206
  241. *
  242. * @param \GearmanJob $job
  243. * @param mixed $context
  244. *
  245. * @return mixed
  246. */
  247. public function handleJob(\GearmanJob $job, $context)
  248. {
  249. if (
  250. !is_array($context)
  251. || !array_key_exists('job_object_instance', $context)
  252. || !array_key_exists('job_method', $context)
  253. ) {
  254. }
  255. $result = call_user_func_array(
  256. array($context['job_object_instance'], $context['job_method']),
  257. array($job, $context)
  258. );
  259. /**
  260. * Workaround for PECL bug #17114
  261. * http://pecl.php.net/bugs/bug.php?id=17114
  262. */
  263. $type = gettype($result);
  264. settype($result, $type);
  265. return $result;
  266. }
  267. }