QueueController::getStreamFunction() - Code Metrics - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

QueueController::getStreamFunction() A
last analyzed 2025-06-22 22:56 UTC

↳ Parent: QueueController

Complexity

Conditions	5
Paths	1

Size

Total Lines	34
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	14.1113

Importance

Changes

Metric	Value
cc	5
eloc	27
nc	1
nop	2
dl	0
loc	34
ccs	8
cts	28
cp	0.2857
crap	14.1113
rs	9.1768
c	0
b	0
f	0

<?php

namespace Dtc\QueueBundle\Controller;

use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
use Dtc\QueueBundle\Exception\UnsupportedException;
use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Model\Worker;
use Symfony\Component\DependencyInjection\ContainerInterface;
filter:
    dependency_paths: ["lib/*"]
use Symfony\Component\HttpFoundation\Request;
filter:
    dependency_paths: ["lib/*"]
use Symfony\Component\HttpFoundation\StreamedResponse;
filter:
    dependency_paths: ["lib/*"]

class QueueController
{
    use ControllerTrait;

    private $container;

    public function __construct(ContainerInterface $container)
    {
        $this->container = $container;
    }

    /**
     * Summary stats.
     */
    public function status()
    {
        $params = [];
        $jobManager = $this->container->get('dtc_queue.manager.job');

        $params['status'] = $jobManager->getStatus();
        $this->addCssJs($params);

        return $this->render('@DtcQueue/Queue/status.html.twig', $params);
    }

    /**
     * List jobs in system by default.
     *
     * @throws UnsupportedException|\Exception
     */
    public function jobsAll()
    {
        $this->validateManagerType('dtc_queue.manager.job');
        $this->checkDtcGridBundle();

        $class1 = $this->container->getParameter('dtc_queue.class.job');
        $class2 = $this->container->getParameter('dtc_queue.class.job_archive');
        $label1 = 'Non-Archived Jobs';
        $label2 = 'Archived Jobs';

        $params = $this->getDualGridParams($class1, $class2, $label1, $label2);

        return $this->render('@DtcQueue/Queue/grid.html.twig', $params);
    }

    /**
     * @throws UnsupportedException
     */
    public function archive(Request $request)
    {
        return $this->streamResults($request, 'archiveAllJobs');
    }

    /**
     * @return StreamedResponse
     *
     * @throws UnsupportedException
     */
    public function resetStalled(Request $request)
    {
        return $this->streamResults($request, 'resetStalledJobs');
    }

    /**
     * @return StreamedResponse
     *
     * @throws UnsupportedException
     */
    public function pruneStalled(Request $request)
    {
        return $this->streamResults($request, 'pruneStalledJobs');
    }

    /**
     * @param $functionName
     *
     * @return StreamedResponse
     *
     * @throws UnsupportedException
     */
    protected function streamResults(Request $request, $functionName)
    {
        $jobManager = $this->container->get('dtc_queue.manager.job');
        if (!$jobManager instanceof DoctrineJobManager) {
            throw new UnsupportedException('$jobManager must be instance of '.DoctrineJobManager::class);
        }

        $streamingResponse = new StreamedResponse($this->getStreamFunction($request, $functionName));
        $streamingResponse->headers->set('Content-Type', 'application/x-ndjson');
        $streamingResponse->headers->set('X-Accel-Buffering', 'no');

        return $streamingResponse;
    }

    /**
     * @param string $functionName
     *
     * @return \Closure
     */
    protected function getStreamFunction(Request $request, $functionName)
    {
        $jobManager = $this->container->get('dtc_queue.manager.job');
        $workerName = $request->get('workerName');
        $methodName = $request->get('method');
        $total = null;
        $callback = function ($count, $totalCount) use (&$total) {
            if (null !== $totalCount && null === $total) {
                $total = $totalCount;
                echo json_encode(['total' => $total]);
                echo "\n";
                flush();

                return;
            }
            echo json_encode(['count' => $count]);
            echo "\n";
            flush();
        };

        return function () use ($jobManager, $callback, $workerName, $methodName, $functionName, &$total) {
            switch ($functionName) {
                case 'archiveAllJobs':
                    $total = $jobManager->countLiveJobs($workerName, $methodName);
                    echo json_encode(['total' => $total]);
                    echo "\n";
                    flush();
                    if ($total > 0) {
                        $jobManager->archiveAllJobs($workerName, $methodName, $callback);
                    }
                    break;
                default:
                    $jobManager->$functionName($workerName, $methodName, $callback);
                    break;
            }
        };
    }

    /**
     * List jobs in system by default.
     *
     * @throws UnsupportedException|\Exception
     */
    public function jobs()
    {
        $this->validateManagerType('dtc_queue.manager.job');
        $this->checkDtcGridBundle();
        $managerType = $this->container->getParameter('dtc_queue.manager.job');
        $rendererFactory = $this->container->get('dtc_grid.renderer.factory');
        $renderer = $rendererFactory->create('datatables');
        $gridSource = $this->container->get('dtc_queue.grid_source.jobs_waiting.'.('mongodb' === $managerType ? 'odm' : $managerType));
        $renderer->bind($gridSource);
        $params = $renderer->getParams();
        $this->addCssJs($params);

        $params['worker_methods'] = $this->container->get('dtc_queue.manager.job')->getWorkersAndMethods();
        $params['prompt_message'] = 'This will archive all non-running jobs';

        return $this->render('@DtcQueue/Queue/jobs.html.twig', $params);
    }

    /**
     * List jobs in system by default.
     *
     * @throws UnsupportedException|\Exception
     */
    public function jobsRunning()
    {
        $this->validateManagerType('dtc_queue.manager.job');
        $this->checkDtcGridBundle();
        $managerType = $this->container->getParameter('dtc_queue.manager.job');
        $rendererFactory = $this->container->get('dtc_grid.renderer.factory');
        $renderer = $rendererFactory->create('datatables');
        $gridSource = $this->container->get('dtc_queue.grid_source.jobs_running.'.('mongodb' === $managerType ? 'odm' : $managerType));
        $renderer->bind($gridSource);
        $params = $renderer->getParams();
        $this->addCssJs($params);

        $params['worker_methods'] = $this->container->get('dtc_queue.manager.job')->getWorkersAndMethods(BaseJob::STATUS_RUNNING);
        $params['prompt_message'] = 'This will prune all stalled jobs';

        return $this->render('@DtcQueue/Queue/jobs_running.html.twig', $params);
    }

    /**
     * @param string $class1
     * @param string $class2
     * @param string $label1
     * @param string $label2
     *
     * @return array
     *
     * @throws \Exception
     */
    protected function getDualGridParams($class1, $class2, $label1, $label2)
    {
        $rendererFactory = $this->container->get('dtc_grid.renderer.factory');
        $renderer = $rendererFactory->create('datatables');
        $gridSource = $this->container->get('dtc_grid.manager.source')->get($class1);
        $renderer->bind($gridSource);
        $params = $renderer->getParams();

        $renderer2 = $rendererFactory->create('datatables');
        $gridSource = $this->container->get('dtc_grid.manager.source')->get($class2);
        $renderer2->bind($gridSource);
        $params2 = $renderer2->getParams();

        $params['archive_grid'] = $params2['dtc_grid'];

        $params['dtc_queue_grid_label1'] = $label1;
        $params['dtc_queue_grid_label2'] = $label2;
        $this->addCssJs($params);

        return $params;
    }

    /**
     * List jobs in system by default.
     *
     * @throws UnsupportedException|\Exception
     */
    public function runs()
    {
        $this->validateRunManager();
        $this->checkDtcGridBundle();
        $class1 = $this->container->getParameter('dtc_queue.class.run');
        $class2 = $this->container->getParameter('dtc_queue.class.run_archive');
        $label1 = 'Live Runs';
        $label2 = 'Archived Runs';

        $params = $this->getDualGridParams($class1, $class2, $label1, $label2);

        return $this->render('@DtcQueue/Queue/grid.html.twig', $params);
    }

    /**
     * List registered workers in the system.
     */
    public function workers()
    {
        $workerManager = $this->container->get('dtc_queue.manager.worker');
        $workers = $workerManager->getWorkers();

        $workerList = [];
        foreach ($workers as $workerName => $worker) {
            /* @var Worker $worker */
            $workerList[$workerName] = get_class($worker);
        }
        $params = ['workers' => $workerList];
        $this->addCssJs($params);

        return $this->render('@DtcQueue/Queue/workers.html.twig', $params);
    }

    /**
     * Validates that DtcGridBundle exists.
     *
     * @throws UnsupportedException
     */
    protected function checkDtcGridBundle()
    {
        if (!class_exists('Dtc\GridBundle\DtcGridBundle')) {
            throw new UnsupportedException('DtcGridBundle (mmucklo/grid-bundle) needs to be installed.');
        }
    }
}


1		<?php
2
3		namespace Dtc\QueueBundle\Controller;
4
5		use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
6		use Dtc\QueueBundle\Exception\UnsupportedException;
7		use Dtc\QueueBundle\Model\BaseJob;
8		use Dtc\QueueBundle\Model\Worker;
9		use Symfony\Component\DependencyInjection\ContainerInterface;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `Symfony\Component\Depend...tion\ContainerInterface` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
10		use Symfony\Component\HttpFoundation\Request;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `Symfony\Component\HttpFoundation\Request` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
11		use Symfony\Component\HttpFoundation\StreamedResponse;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `Symfony\Component\HttpFoundation\StreamedResponse` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
12
13		class QueueController
14		{
15	1	use ControllerTrait;
16
17		private $container;
18
19	7	public function __construct(ContainerInterface $container)
20		{
21	7	$this->container = $container;
22	7	}
23
24		/**
25		* Summary stats.
26		*/
27	1	public function status()
28		{
29	1	$params = [];
30	1	$jobManager = $this->container->get('dtc_queue.manager.job');
31
32	1	$params['status'] = $jobManager->getStatus();
33	1	$this->addCssJs($params);
34
35	1	return $this->render('@DtcQueue/Queue/status.html.twig', $params);
36		}
37
38		/**
39		* List jobs in system by default.
40		*
41		* @throws UnsupportedException\|\Exception
42		*/
43	1	public function jobsAll()
44		{
45	1	$this->validateManagerType('dtc_queue.manager.job');
46	1	$this->checkDtcGridBundle();
47
48	1	$class1 = $this->container->getParameter('dtc_queue.class.job');
49	1	$class2 = $this->container->getParameter('dtc_queue.class.job_archive');
50	1	$label1 = 'Non-Archived Jobs';
51	1	$label2 = 'Archived Jobs';
52
53	1	$params = $this->getDualGridParams($class1, $class2, $label1, $label2);
54
55	1	return $this->render('@DtcQueue/Queue/grid.html.twig', $params);
56		}
57
58		/**
59		* @throws UnsupportedException
60		*/
61	1	public function archive(Request $request)
62		{
63	1	return $this->streamResults($request, 'archiveAllJobs');
64		}
65
66		/**
67		* @return StreamedResponse
68		*
69		* @throws UnsupportedException
70		*/
71		public function resetStalled(Request $request)
72		{
73		return $this->streamResults($request, 'resetStalledJobs');
74		}
75
76		/**
77		* @return StreamedResponse
78		*
79		* @throws UnsupportedException
80		*/
81		public function pruneStalled(Request $request)
82		{
83		return $this->streamResults($request, 'pruneStalledJobs');
84		}
85
86		/**
87		* @param $functionName
88		*
89		* @return StreamedResponse
90		*
91		* @throws UnsupportedException
92		*/
93	1	protected function streamResults(Request $request, $functionName)
94		{
95	1	$jobManager = $this->container->get('dtc_queue.manager.job');
96	1	if (!$jobManager instanceof DoctrineJobManager) {
97		throw new UnsupportedException('$jobManager must be instance of '.DoctrineJobManager::class);
98		}
99
100	1	$streamingResponse = new StreamedResponse($this->getStreamFunction($request, $functionName));
101	1	$streamingResponse->headers->set('Content-Type', 'application/x-ndjson');
102	1	$streamingResponse->headers->set('X-Accel-Buffering', 'no');
103
104	1	return $streamingResponse;
105		}
106
107		/**
108		* @param string $functionName
109		*
110		* @return \Closure
111		*/
112	1	protected function getStreamFunction(Request $request, $functionName)
113		{
114	1	$jobManager = $this->container->get('dtc_queue.manager.job');
115	1	$workerName = $request->get('workerName');
116	1	$methodName = $request->get('method');
117	1	$total = null;
118	1	$callback = function ($count, $totalCount) use (&$total) {
119		if (null !== $totalCount && null === $total) {
120		$total = $totalCount;
121		echo json_encode(['total' => $total]);
122		echo "\n";
123		flush();
124
125		return;
126		}
127		echo json_encode(['count' => $count]);
128		echo "\n";
129		flush();
130	1	};
131
132	1	return function () use ($jobManager, $callback, $workerName, $methodName, $functionName, &$total) {
133		switch ($functionName) {
134		case 'archiveAllJobs':
135		$total = $jobManager->countLiveJobs($workerName, $methodName);
136		echo json_encode(['total' => $total]);
137		echo "\n";
138		flush();
139		if ($total > 0) {
140		$jobManager->archiveAllJobs($workerName, $methodName, $callback);
141		}
142		break;
143		default:
144		$jobManager->$functionName($workerName, $methodName, $callback);
145		break;
146		}
147	1	};
148		}
149
150		/**
151		* List jobs in system by default.
152		*
153		* @throws UnsupportedException\|\Exception
154		*/
155	1	public function jobs()
156		{
157	1	$this->validateManagerType('dtc_queue.manager.job');
158	1	$this->checkDtcGridBundle();
159	1	$managerType = $this->container->getParameter('dtc_queue.manager.job');
160	1	$rendererFactory = $this->container->get('dtc_grid.renderer.factory');
161	1	$renderer = $rendererFactory->create('datatables');
162	1	$gridSource = $this->container->get('dtc_queue.grid_source.jobs_waiting.'.('mongodb' === $managerType ? 'odm' : $managerType));
163	1	$renderer->bind($gridSource);
164	1	$params = $renderer->getParams();
165	1	$this->addCssJs($params);
166
167	1	$params['worker_methods'] = $this->container->get('dtc_queue.manager.job')->getWorkersAndMethods();
168	1	$params['prompt_message'] = 'This will archive all non-running jobs';
169
170	1	return $this->render('@DtcQueue/Queue/jobs.html.twig', $params);
171		}
172
173		/**
174		* List jobs in system by default.
175		*
176		* @throws UnsupportedException\|\Exception
177		*/
178	1	public function jobsRunning()
179		{
180	1	$this->validateManagerType('dtc_queue.manager.job');
181	1	$this->checkDtcGridBundle();
182	1	$managerType = $this->container->getParameter('dtc_queue.manager.job');
183	1	$rendererFactory = $this->container->get('dtc_grid.renderer.factory');
184	1	$renderer = $rendererFactory->create('datatables');
185	1	$gridSource = $this->container->get('dtc_queue.grid_source.jobs_running.'.('mongodb' === $managerType ? 'odm' : $managerType));
186	1	$renderer->bind($gridSource);
187	1	$params = $renderer->getParams();
188	1	$this->addCssJs($params);
189
190	1	$params['worker_methods'] = $this->container->get('dtc_queue.manager.job')->getWorkersAndMethods(BaseJob::STATUS_RUNNING);
191	1	$params['prompt_message'] = 'This will prune all stalled jobs';
192
193	1	return $this->render('@DtcQueue/Queue/jobs_running.html.twig', $params);
194		}
195
196		/**
197		* @param string $class1
198		* @param string $class2
199		* @param string $label1
200		* @param string $label2
201		*
202		* @return array
203		*
204		* @throws \Exception
205		*/
206	2	protected function getDualGridParams($class1, $class2, $label1, $label2)
207		{
208	2	$rendererFactory = $this->container->get('dtc_grid.renderer.factory');
209	2	$renderer = $rendererFactory->create('datatables');
210	2	$gridSource = $this->container->get('dtc_grid.manager.source')->get($class1);
211	2	$renderer->bind($gridSource);
212	2	$params = $renderer->getParams();
213
214	2	$renderer2 = $rendererFactory->create('datatables');
215	2	$gridSource = $this->container->get('dtc_grid.manager.source')->get($class2);
216	2	$renderer2->bind($gridSource);
217	2	$params2 = $renderer2->getParams();
218
219	2	$params['archive_grid'] = $params2['dtc_grid'];
220
221	2	$params['dtc_queue_grid_label1'] = $label1;
222	2	$params['dtc_queue_grid_label2'] = $label2;
223	2	$this->addCssJs($params);
224
225	2	return $params;
226		}
227
228		/**
229		* List jobs in system by default.
230		*
231		* @throws UnsupportedException\|\Exception
232		*/
233	1	public function runs()
234		{
235	1	$this->validateRunManager();
236	1	$this->checkDtcGridBundle();
237	1	$class1 = $this->container->getParameter('dtc_queue.class.run');
238	1	$class2 = $this->container->getParameter('dtc_queue.class.run_archive');
239	1	$label1 = 'Live Runs';
240	1	$label2 = 'Archived Runs';
241
242	1	$params = $this->getDualGridParams($class1, $class2, $label1, $label2);
243
244	1	return $this->render('@DtcQueue/Queue/grid.html.twig', $params);
245		}
246
247		/**
248		* List registered workers in the system.
249		*/
250	1	public function workers()
251		{
252	1	$workerManager = $this->container->get('dtc_queue.manager.worker');
253	1	$workers = $workerManager->getWorkers();
254
255	1	$workerList = [];
256	1	foreach ($workers as $workerName => $worker) {
257		/* @var Worker $worker */
258		$workerList[$workerName] = get_class($worker);
259		}
260	1	$params = ['workers' => $workerList];
261	1	$this->addCssJs($params);
262
263	1	return $this->render('@DtcQueue/Queue/workers.html.twig', $params);
264		}
265
266		/**
267		* Validates that DtcGridBundle exists.
268		*
269		* @throws UnsupportedException
270		*/
271	4	protected function checkDtcGridBundle()
272		{
273	4	if (!class_exists('Dtc\GridBundle\DtcGridBundle')) {
274		throw new UnsupportedException('DtcGridBundle (mmucklo/grid-bundle) needs to be installed.');
275		}
276	4	}
277		}
278

mmucklo / DtcQueueBundle

QueueController::getStreamFunction() A last analyzed 2025-06-22 22:56 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

QueueController::getStreamFunction() A
last analyzed 2025-06-22 22:56 UTC