hazza/Phraseanet
1
0
Fork 0
mirror of https://github.com/alchemy-fr/Phraseanet.git synced 2025-10-23 09:53:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

add tab queue monitor, limit job list to 500

Browse Source
This commit is contained in:
aynsix
2020-06-10 19:25:16 +03:00
parent b8487cd141
commit 6d31f93774
15 changed files with 2340 additions and 1794 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/Alchemy/Phrasea/Model/Entities/WorkerRunningJob.php
View File

@@ -20,6 +20,8 @@ class WorkerRunningJob
const FINISHED = 'finished';
const RUNNING = 'running';
const MAX_RESULT = 500;
/**
* @ORM\Column(type="integer")
* @ORM\Id

22
lib/Alchemy/Phrasea/WorkerManager/Controller/AdminConfigurationController.php
View File

@@ -79,11 +79,11 @@ class AdminConfigurationController extends Controller
$reload = ($request->query->get('reload')) == 1 ? true : false ;
if ($request->query->get('running') == 1 && $request->query->get('finished') == 1) {
$workerRunningJob = $repoWorker->findAll();
$workerRunningJob = $repoWorker->findBy([], ['id' => 'DESC'], WorkerRunningJob::MAX_RESULT);
} elseif ($request->query->get('running') == 1) {
$workerRunningJob = $repoWorker->findBy(['status' => WorkerRunningJob::RUNNING]);
$workerRunningJob = $repoWorker->findBy(['status' => WorkerRunningJob::RUNNING], ['id' => 'DESC'], WorkerRunningJob::MAX_RESULT);
} elseif ($request->query->get('finished') == 1) {
$workerRunningJob = $repoWorker->findBy(['status' => WorkerRunningJob::FINISHED]);
$workerRunningJob = $repoWorker->findBy(['status' => WorkerRunningJob::FINISHED], ['id' => 'DESC'], WorkerRunningJob::MAX_RESULT);
}
return $this->render('admin/worker-manager/worker_info.html.twig', [
@@ -92,6 +92,22 @@ class AdminConfigurationController extends Controller
]);
}
public function queueMonitorAction(PhraseaApplication $app, Request $request)
{
$reload = ($request->query->get('reload')) == 1 ? true : false ;
/** @var AMQPConnection $serverConnection */
$serverConnection = $app['alchemy_worker.amqp.connection'];
$serverConnection->getChannel();
$serverConnection->declareExchange();
$queuesStatus = $serverConnection->getQueuesStatus();
return $this->render('admin/worker-manager/worker_queue_monitor.html.twig', [
'queuesStatus' => $queuesStatus,
'reload' => $reload
]);
}
public function truncateTableAction(PhraseaApplication $app, Request $request)
{
/** @var WorkerRunningJobRepository $repoWorker */

4
lib/Alchemy/Phrasea/WorkerManager/Provider/ControllerServiceProvider.php
View File

@@ -84,6 +84,10 @@ class ControllerServiceProvider implements ControllerProviderInterface, ServiceP
->method('GET|POST')
->bind('worker_admin_pullAssets');
$controllers->match('/queue-monitor', 'controller.worker.admin.configuration:queueMonitorAction')
->method('GET')
->bind('worker_admin_queue_monitor');
return $controllers;
}

31
lib/Alchemy/Phrasea/WorkerManager/Queue/AMQPConnection.php
View File

@@ -213,6 +213,37 @@ class AMQPConnection
}
}
/**
* Get queueName, messageCount, consumerCount of queues
* @return array
*/
public function getQueuesStatus()
{
$queuesList = array_merge(
array_values(self::$defaultQueues),
array_values(self::$defaultDelayedQueues),
array_values(self::$defaultRetryQueues),
array_values(self::$defaultFailedQueues)
);
$this->getChannel();
$queuesStatus = [];
foreach ($queuesList as $queue) {
$this->setQueue($queue);
list($queueName, $messageCount, $consumerCount) = $this->channel->queue_declare($queue, true);
$status['queueName'] = $queueName;
$status['messageCount'] = $messageCount;
$status['consumerCount'] = $consumerCount;
$queuesStatus[] = $status;
unset($status);
}
return $queuesStatus;
}
public function connectionClose()
{
$this->channel->close();