Worker::__construct() - Code Metrics - Inspection of "GearmanJob->sendComplete() - Object error (#7)" - dominikmank/gearman - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 329e1f...6c2475 )

by Dominik

created 2017-08-28 10:22 UTC

Worker::__construct() A

↳ Parent: Worker

Complexity

Conditions	1
Paths	1

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes

Metric	Value
dl	0
loc	9
ccs	5
cts	5
cp	1
rs	9.6666
c	0
b	0
f	0
cc	1
eloc	7
nc	1
nop	3
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;

    /**
     * @var bool
     */
    private $killRequested = false;

    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 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, $result)
                        );
                    } catch (\Exception $e) {

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

                        $result = $e;
                    }

                    if (is_object($result) || is_array($result)) {
                        $result = serialize($result);
                    }

                    $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
        ) && !$this->killRequested;
    }

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

        $this->killRequested = true;
    }
}


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		/**
44		* @var bool
45		*/
46		private $killRequested = false;
47
48	3	public function __construct(
49		ServerCollection $servers,
50		JobCollection $jobs,
51		EventDispatcherInterface $dispatcher
52		) {
53	3	$this->serverCollection = $servers;
54	3	$this->jobs = $jobs;
55	3	$this->dispatcher = $dispatcher;
56	3	}
57
58		/**
59		* @param \GearmanWorker $worker
60		*/
61	3	public function setImplementation(\GearmanWorker $worker)
62		{
63	3	$this->realWorker = $worker;
64	3	}
65
66	3	public function run()
67		{
68	3	$this->getEventDispatcher()->dispatch(
69	3	WorkerEvent::EVENT_BEFORE_RUN,
70	3	new WorkerEvent($this)
71	3	);
72
73		try {
74
75	3	while ($this->workerIsPending()) {
76
77		$gearmanReturnCode = $this->getWorker()->returnCode();
78
79		if ($gearmanReturnCode == GEARMAN_IO_WAIT) {
80
81		$this->getEventDispatcher()->dispatch(
82		WorkerEvent::EVENT_ON_IO_WAIT,
83		new WorkerEvent($this)
84		);
85
86		@$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...
87		}
88
89		if ($gearmanReturnCode == GEARMAN_NO_JOBS) {
90
91		$this->getEventDispatcher()->dispatch(
92		WorkerEvent::EVENT_ON_NO_JOBS,
93		new WorkerEvent($this)
94		);
95
96		}
97
98		if ($gearmanReturnCode != GEARMAN_SUCCESS) {
99
100		$this->getEventDispatcher()->dispatch(
101		WorkerEvent::EVENT_ON_WORK,
102		new WorkerEvent($this)
103		);
104		}
105
106		if ($gearmanReturnCode == GEARMAN_SUCCESS) {
107
108		$this->getEventDispatcher()->dispatch(
109		WorkerEvent::EVENT_AFTER_RUN,
110		new WorkerEvent($this)
111		);
112		}
113
114		}
115	3	} catch (NoFunctionGiven $e) {
116	1	throw $e;
117		} catch (\Exception $e) {
118		$this->getEventDispatcher()->dispatch(
119		WorkerExceptionEvent::EVENT_ON_FAILURE,
120		new WorkerExceptionEvent($this, $e)
121		);
122
123		}
124	2	}
125
126		/**
127		* @return EventDispatcherInterface
128		*/
129	3	protected function getEventDispatcher()
130		{
131	3	return $this->dispatcher;
132		}
133
134		/**
135		* @return \GearmanWorker
136		*/
137	3	private function getWorker()
138		{
139	3	if (!$this->realWorker) {
140		$this->realWorker = new \GearmanWorker();
141		$this->realWorker->setOptions(GEARMAN_WORKER_NON_BLOCKING);
142		}
143
144	3	if (!$this->connectedToServer) {
145
146	3	$this->getEventDispatcher()->dispatch(
147	3	ConnectToServerEvent::CONNECT_TO_SERVER_EVENT,
148	3	new ConnectToServerEvent($this->serverCollection)
149	3	);
150
151		/* @var \dmank\gearman\Server $server */
152	3	foreach ($this->serverCollection->getServers() as $server) {
153	3	$this->realWorker->addServer($server->getHost(), $server->getPort());
154	3	}
155
156	3	$this->getEventDispatcher()->dispatch(
157	3	ConnectToServerEvent::CONNECTED_TO_SERVER_EVENT,
158	3	new ConnectToServerEvent($this->serverCollection)
159	3	);
160
161	3	$this->connectedToServer = true;
162	3	}
163
164	3	if (!$this->registeredFunctions) {
165	3	$this->registerFunctions();
166
167	2	$this->registeredFunctions = true;
168	2	}
169
170	2	return $this->realWorker;
171		}
172
173	3	private function registerFunctions()
174		{
175	3	$this->getEventDispatcher()->dispatch(
176	3	RegisterFunctionEvent::EVENT_ON_BEFORE_REGISTER_FUNCTIONS,
177	3	new RegisterFunctionEvent($this->jobs)
178	3	);
179
180	3	if (count($this->jobs) == 0) {
181
182	1	throw new NoFunctionGiven(
183	1	sprintf('Didnt have jobs to register. So we need to stop here. My bad!')
184	1	);
185		}
186
187		/* @var JobHandlerInterface $jobClass */
188	2	foreach ($this->jobs->getJobs() as $jobName => $jobClass) {
189	2	$this->realWorker->addFunction(
190	2	$jobName,
191		function (\GearmanJob $gearmanJob) use ($jobClass, $jobName) {
192
193		$job = new Job($jobName, $gearmanJob->workload());
194
195		try {
196
197		$this->getEventDispatcher()->dispatch(
198		FunctionEvent::FUNCTION_BEFORE_EXECUTE,
199		new FunctionEvent($jobClass, $job)
200		);
201
202		$result = $jobClass->execute($job);
203
204		$this->getEventDispatcher()->dispatch(
205		FunctionEvent::FUNCTION_AFTER_EXECUTE,
206		new FunctionEvent($jobClass, $job, $result)
207		);
208		} catch (\Exception $e) {
209
210		$this->getEventDispatcher()->dispatch(
211		FunctionFailureEvent::FUNCTION_ON_FAILURE,
212		new FunctionFailureEvent($jobClass, $e, $job)
213		);
214
215		$result = $e;
216		}
217
218		if (is_object($result) \|\| is_array($result)) {
219		$result = serialize($result);
220		}
221
222		$gearmanJob->sendComplete($result);
223		}
224	2	);
225	2	}
226
227	2	$this->getEventDispatcher()->dispatch(
228	2	RegisterFunctionEvent::EVENT_ON_AFTER_REGISTER_FUNCTIONS,
229	2	new RegisterFunctionEvent($this->jobs)
230	2	);
231
232	2	}
233
234		/**
235		* @return bool
236		*/
237	3	private function workerIsPending()
238		{
239		return (
240	3	$this->getWorker()->work() \|\|
241	2	$this->getWorker()->returnCode() == GEARMAN_IO_WAIT \|\|
242	2	$this->getWorker()->returnCode() == GEARMAN_NO_JOBS
243	2	) && !$this->killRequested;
244		}
245
246		public function destroyWorker()
247		{
248		$this->getEventDispatcher()->dispatch(
249		WorkerEvent::EVENT_ON_BEFORE_DESTROY,
250		new WorkerEvent($this)
251		);
252
253		$this->killRequested = true;
254		}
255		}
256

dominikmank / gearman

GitHub Access Token became invalid

Push — master ( 329e1f...6c2475 )

Worker::__construct() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like