Worker::setImplementation() - Code Metrics - Inspection of "- composer revert phpunit to ^4 for php 5.5 suppor..." - dominikmank/gearman - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6f49e2...e22812 )

by Dominik

created 2017-04-27 07:07 UTC

Worker::setImplementation() A

↳ Parent: Worker

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
dl	0
loc	4
ccs	3
cts	3
cp	1
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	1
crap	1

<?php

namespace dmank\gearman;

use dmank\gearman\event\ConnectToServerEvent;
use dmank\gearman\event\FunctionEvent;
use dmank\gearman\event\FunctionFailureEvent;
use dmank\gearman\event\RegisterFunctionEvent;
use dmank\gearman\event\WorkerEvent;
use dmank\gearman\event\WorkerExceptionEvent;
use dmank\gearman\exception\NoFunctionGiven;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

class Worker
{
    /**
     * @var \GearmanWorker
     */
    private $realWorker;
    /**
     * @var bool
     */
    private $connectedToServer = false;
    /**
     * @var bool
     */
    private $registeredFunctions = false;
    /**
     * @var ServerCollection
     */
    private $serverCollection;

    /**
     * @var JobCollection
     */
    private $jobs = [];

    /**
     * @var EventDispatcherInterface
     */
    private $dispatcher;

    public function __construct(
        ServerCollection $servers,
        JobCollection $jobs,
        EventDispatcherInterface $dispatcher
    )
    {
        $this->serverCollection = $servers;
        $this->jobs = $jobs;
        $this->dispatcher = $dispatcher;
    }

    /**
     * @param \GearmanWorker $worker
     */
    public function setImplementation(\GearmanWorker $worker)
    {
        $this->realWorker = $worker;
    }

    public function run()
    {
        $this->getEventDispatcher()->dispatch(
            WorkerEvent::EVENT_BEFORE_RUN,
            new WorkerEvent($this)
        );

        try {

            while ($this->workerIsPending()) {

                $gearmanReturnCode = $this->getWorker()->returnCode();

                if ($gearmanReturnCode == GEARMAN_IO_WAIT) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_IO_WAIT,
                        new WorkerEvent($this)
                    );

                    @$this->getWorker()->wait();
// For example instead of
@mkdir($dir);

// Better use
if (@mkdir($dir) === false) {
    throw new \RuntimeException('The directory '.$dir.' could not be created.');
}
                }

                if ($gearmanReturnCode == GEARMAN_NO_JOBS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_NO_JOBS,
                        new WorkerEvent($this)
                    );

                }

                if ($gearmanReturnCode != GEARMAN_SUCCESS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_ON_WORK,
                        new WorkerEvent($this)
                    );
                }

                if ($gearmanReturnCode == GEARMAN_SUCCESS) {

                    $this->getEventDispatcher()->dispatch(
                        WorkerEvent::EVENT_AFTER_RUN,
                        new WorkerEvent($this)
                    );
                }

            }
        } catch (NoFunctionGiven $e) {
            throw $e;
        } catch (\Exception $e) {
            $this->getEventDispatcher()->dispatch(
                WorkerExceptionEvent::EVENT_ON_FAILURE,
                new WorkerExceptionEvent($this, $e)
            );

        }

        return;
    }

    /**
     * @return EventDispatcherInterface
     */
    protected function getEventDispatcher()
    {
        return $this->dispatcher;
    }

    /**
     * @return \GearmanWorker
     */
    private function getWorker()
    {
        if (!$this->realWorker) {
            $this->realWorker = new \GearmanWorker();
            $this->realWorker->setOptions(GEARMAN_WORKER_NON_BLOCKING);
        }

        if (!$this->connectedToServer) {

            $this->getEventDispatcher()->dispatch(
                ConnectToServerEvent::CONNECT_TO_SERVER_EVENT,
                new ConnectToServerEvent($this->serverCollection)
            );

            /* @var \dmank\gearman\Server $server */
            foreach ($this->serverCollection->getServers() as $server) {
                $this->realWorker->addServer($server->getHost(), $server->getPort());
            }

            $this->getEventDispatcher()->dispatch(
                ConnectToServerEvent::CONNECTED_TO_SERVER_EVENT,
                new ConnectToServerEvent($this->serverCollection)
            );

            $this->connectedToServer = true;
        }

        if (!$this->registeredFunctions) {
            $this->registerFunctions();

            $this->registeredFunctions = true;
        }

        return $this->realWorker;
    }

    private function registerFunctions()
    {
        $this->getEventDispatcher()->dispatch(
            RegisterFunctionEvent::EVENT_ON_BEFORE_REGISTER_FUNCTIONS,
            new RegisterFunctionEvent($this->jobs)
        );

        if (count($this->jobs) == 0) {

            throw new NoFunctionGiven(
                sprintf('Didnt have jobs to register. So we need to stop here. My bad!')
            );
        }

        /* @var JobHandlerInterface $jobClass */
        foreach ($this->jobs->getJobs() as $jobName => $jobClass) {
            $this->realWorker->addFunction(
                $jobName,
                function (\GearmanJob $gearmanJob) use ($jobClass, $jobName) {

                    $job = new Job($jobName, $gearmanJob->workload());

                    try {

                        $this->getEventDispatcher()->dispatch(
                            FunctionEvent::FUNCTION_BEFORE_EXECUTE,
                            new FunctionEvent($jobClass, $job)
                        );

                        $result = $jobClass->execute($job);

                        $this->getEventDispatcher()->dispatch(
                            FunctionEvent::FUNCTION_AFTER_EXECUTE,
                            new FunctionEvent($jobClass, $job)
                        );
                    } catch (\Exception $e) {

                        $this->getEventDispatcher()->dispatch(
                            FunctionFailureEvent::FUNCTION_ON_FAILURE,
                            new FunctionFailureEvent($jobClass, $e, $job)
                        );

                        $result = $e;
                    }

                    $gearmanJob->sendComplete($result);
                }
            );
        }

        $this->getEventDispatcher()->dispatch(
            RegisterFunctionEvent::EVENT_ON_AFTER_REGISTER_FUNCTIONS,
            new RegisterFunctionEvent($this->jobs)
        );

    }

    /**
     * @return bool
     */
    private function workerIsPending()
    {
        return (
            $this->getWorker()->work() ||
            $this->getWorker()->returnCode() == GEARMAN_IO_WAIT ||
            $this->getWorker()->returnCode() == GEARMAN_NO_JOBS
        );
    }

    public function destroyWorker()
    {
        $this->getEventDispatcher()->dispatch(
            WorkerEvent::EVENT_ON_BEFORE_DESTROY,
            new WorkerEvent($this)
        );

        unset($this->realWorker);
        $this->realWorker = null;

        posix_kill(posix_getpid(), SIGTERM);
    }
}


1		<?php
2
3		namespace dmank\gearman;
4
5		use dmank\gearman\event\ConnectToServerEvent;
6		use dmank\gearman\event\FunctionEvent;
7		use dmank\gearman\event\FunctionFailureEvent;
8		use dmank\gearman\event\RegisterFunctionEvent;
9		use dmank\gearman\event\WorkerEvent;
10		use dmank\gearman\event\WorkerExceptionEvent;
11		use dmank\gearman\exception\NoFunctionGiven;
12		use Symfony\Component\EventDispatcher\EventDispatcherInterface;
13
14		class Worker
15		{
16		/**
17		* @var \GearmanWorker
18		*/
19		private $realWorker;
20		/**
21		* @var bool
22		*/
23		private $connectedToServer = false;
24		/**
25		* @var bool
26		*/
27		private $registeredFunctions = false;
28		/**
29		* @var ServerCollection
30		*/
31		private $serverCollection;
32
33		/**
34		* @var JobCollection
35		*/
36		private $jobs = [];
37
38		/**
39		* @var EventDispatcherInterface
40		*/
41		private $dispatcher;
42
43	3	public function __construct(
44		ServerCollection $servers,
45		JobCollection $jobs,
46		EventDispatcherInterface $dispatcher
47		)
48		{
49	3	$this->serverCollection = $servers;
50	3	$this->jobs = $jobs;
51	3	$this->dispatcher = $dispatcher;
52	3	}
53
54		/**
55		* @param \GearmanWorker $worker
56		*/
57	3	public function setImplementation(\GearmanWorker $worker)
58		{
59	3	$this->realWorker = $worker;
60	3	}
61
62	3	public function run()
63		{
64	3	$this->getEventDispatcher()->dispatch(
65	3	WorkerEvent::EVENT_BEFORE_RUN,
66	3	new WorkerEvent($this)
67	3	);
68
69		try {
70
71	3	while ($this->workerIsPending()) {
72
73		$gearmanReturnCode = $this->getWorker()->returnCode();
74
75		if ($gearmanReturnCode == GEARMAN_IO_WAIT) {
76
77		$this->getEventDispatcher()->dispatch(
78		WorkerEvent::EVENT_ON_IO_WAIT,
79		new WorkerEvent($this)
80		);
81
82		@$this->getWorker()->wait();
		0 ignored issues – show Security Best Practice introduced 2017-01-16 08:13 UTC by Report Bug Copy Issue Report It seems like you do not handle an error condition here. This can introduce security issues, and is generally not recommended. If you suppress an error, we recommend checking for the error condition explicitly: // For example instead of @mkdir($dir); // Better use if (@mkdir($dir) === false) { throw new \RuntimeException('The directory '.$dir.' could not be created.'); } Loading history...
83		}
84
85		if ($gearmanReturnCode == GEARMAN_NO_JOBS) {
86
87		$this->getEventDispatcher()->dispatch(
88		WorkerEvent::EVENT_ON_NO_JOBS,
89		new WorkerEvent($this)
90		);
91
92		}
93
94		if ($gearmanReturnCode != GEARMAN_SUCCESS) {
95
96		$this->getEventDispatcher()->dispatch(
97		WorkerEvent::EVENT_ON_WORK,
98		new WorkerEvent($this)
99		);
100		}
101
102		if ($gearmanReturnCode == GEARMAN_SUCCESS) {
103
104		$this->getEventDispatcher()->dispatch(
105		WorkerEvent::EVENT_AFTER_RUN,
106		new WorkerEvent($this)
107		);
108		}
109
110		}
111	3	} catch (NoFunctionGiven $e) {
112	1	throw $e;
113		} catch (\Exception $e) {
114		$this->getEventDispatcher()->dispatch(
115		WorkerExceptionEvent::EVENT_ON_FAILURE,
116		new WorkerExceptionEvent($this, $e)
117		);
118
119		}
120
121	2	return;
122		}
123
124		/**
125		* @return EventDispatcherInterface
126		*/
127	3	protected function getEventDispatcher()
128		{
129	3	return $this->dispatcher;
130		}
131
132		/**
133		* @return \GearmanWorker
134		*/
135	3	private function getWorker()
136		{
137	3	if (!$this->realWorker) {
138		$this->realWorker = new \GearmanWorker();
139		$this->realWorker->setOptions(GEARMAN_WORKER_NON_BLOCKING);
140		}
141
142	3	if (!$this->connectedToServer) {
143
144	3	$this->getEventDispatcher()->dispatch(
145	3	ConnectToServerEvent::CONNECT_TO_SERVER_EVENT,
146	3	new ConnectToServerEvent($this->serverCollection)
147	3	);
148
149		/* @var \dmank\gearman\Server $server */
150	3	foreach ($this->serverCollection->getServers() as $server) {
151	3	$this->realWorker->addServer($server->getHost(), $server->getPort());
152	3	}
153
154	3	$this->getEventDispatcher()->dispatch(
155	3	ConnectToServerEvent::CONNECTED_TO_SERVER_EVENT,
156	3	new ConnectToServerEvent($this->serverCollection)
157	3	);
158
159	3	$this->connectedToServer = true;
160	3	}
161
162	3	if (!$this->registeredFunctions) {
163	3	$this->registerFunctions();
164
165	2	$this->registeredFunctions = true;
166	2	}
167
168	2	return $this->realWorker;
169		}
170
171	3	private function registerFunctions()
172		{
173	3	$this->getEventDispatcher()->dispatch(
174	3	RegisterFunctionEvent::EVENT_ON_BEFORE_REGISTER_FUNCTIONS,
175	3	new RegisterFunctionEvent($this->jobs)
176	3	);
177
178	3	if (count($this->jobs) == 0) {
179
180	1	throw new NoFunctionGiven(
181	1	sprintf('Didnt have jobs to register. So we need to stop here. My bad!')
182	1	);
183		}
184
185		/* @var JobHandlerInterface $jobClass */
186	2	foreach ($this->jobs->getJobs() as $jobName => $jobClass) {
187	2	$this->realWorker->addFunction(
188	2	$jobName,
189		function (\GearmanJob $gearmanJob) use ($jobClass, $jobName) {
190
191		$job = new Job($jobName, $gearmanJob->workload());
192
193		try {
194
195		$this->getEventDispatcher()->dispatch(
196		FunctionEvent::FUNCTION_BEFORE_EXECUTE,
197		new FunctionEvent($jobClass, $job)
198		);
199
200		$result = $jobClass->execute($job);
201
202		$this->getEventDispatcher()->dispatch(
203		FunctionEvent::FUNCTION_AFTER_EXECUTE,
204		new FunctionEvent($jobClass, $job)
205		);
206		} catch (\Exception $e) {
207
208		$this->getEventDispatcher()->dispatch(
209		FunctionFailureEvent::FUNCTION_ON_FAILURE,
210		new FunctionFailureEvent($jobClass, $e, $job)
211		);
212
213		$result = $e;
214		}
215
216		$gearmanJob->sendComplete($result);
217		}
218	2	);
219	2	}
220
221	2	$this->getEventDispatcher()->dispatch(
222	2	RegisterFunctionEvent::EVENT_ON_AFTER_REGISTER_FUNCTIONS,
223	2	new RegisterFunctionEvent($this->jobs)
224	2	);
225
226	2	}
227
228		/**
229		* @return bool
230		*/
231	3	private function workerIsPending()
232		{
233		return (
234	3	$this->getWorker()->work() \|\|
235	2	$this->getWorker()->returnCode() == GEARMAN_IO_WAIT \|\|
236	2	$this->getWorker()->returnCode() == GEARMAN_NO_JOBS
237	2	);
238		}
239
240		public function destroyWorker()
241		{
242		$this->getEventDispatcher()->dispatch(
243		WorkerEvent::EVENT_ON_BEFORE_DESTROY,
244		new WorkerEvent($this)
245		);
246
247		unset($this->realWorker);
248		$this->realWorker = null;
249
250		posix_kill(posix_getpid(), SIGTERM);
251		}
252		}
253

dominikmank / gearman

GitHub Access Token became invalid

Push — master ( 6f49e2...e22812 )

Worker::setImplementation() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like