Les queues rendent bien service dès qu'il faut mettre du travail de côté pour le reprendre plus tard. L'idée est simple : nous déposons des tâches ou des données dans une queue (nous créons la queue), puis nous les traitons avec un plugin QueueWorker (nous vidons la queue), généralement sur le cron. Dans ce post, nous allons voir comment faire l'un et l'autre.
Il existe plusieurs façons de remplir une queue :
- depuis un formulaire,
- depuis un contrôleur,
- depuis une implémentation de
hook_cron().
Et plusieurs façons de la traiter :
- sur le cron avec un plugin QueueWorker,
- comme un processus Batch avec un plugin QueueWorker,
- en réclamant chaque élément manuellement dans un service ou un contrôleur.
Ici, nous remplissons la queue depuis un contrôleur et nous la vidons avec un plugin QueueWorker sur le cron. Pourquoi depuis un contrôleur ? Parce qu'on peut alors le déclencher depuis un planificateur externe, une crontab Linux par exemple, au lieu de dépendre du « poor man's cron » de Drupal, qui ne se déclenche qu'aux requêtes de page.
Le module importe le titre et la description de chaque entrée du flux RSS de Drupal Planet, les dépose dans une queue nommée exqueue_import puis, lorsque le cron s'exécute, crée un node de type page par élément. L'exemple original en Drupal 8 se trouve sur github.com/KarimBoudjema/Drupal8-ex-queue-api-01 ; le code ci-dessous est réécrit pour Drupal 11.
Le module comporte deux parties :
- un contrôleur
src/Controller/ExQueueController.phpavec une route dansexqueue01.routing.ymlet deux méthodes :getData()(récupérer les données et les mettre en queue) etdeleteTheQueue()(vider la queue) ; - un plugin QueueWorker
src/Plugin/QueueWorker/ExQueue01.phpqui traite chaque élément.
web/modules/custom/exqueue01/
|-- exqueue01.info.yml
|-- exqueue01.routing.yml
`-- src
|-- Controller
| `-- ExQueueController.php
`-- Plugin
`-- QueueWorker
`-- ExQueue01.phpTout d'abord, générons le squelette du module et de ses plugins avec le générateur de code de Drush 13 (Drupal Console n'existe plus) :
drush generate module
drush generate controller
drush generate plugin:queue-worker1. Remplir la queue depuis un contrôleur
Voici le contrôleur en Drupal 11. Remarquez l'injection de dépendances moderne : la classe utilise le trait de cœur AutowireTrait, ce qui nous évite d'écrire une méthode create() à la main : les attributs #[Autowire] placés sur les propriétés promues du constructeur indiquent au conteneur quels services injecter.
<?php
declare(strict_types=1);
namespace Drupal\exqueue01\Controller;
use Drupal\Core\Controller\ControllerBase;
use Drupal\Core\DependencyInjection\AutowireTrait;
use Drupal\Core\Queue\QueueFactory;
use GuzzleHttp\ClientInterface;
use GuzzleHttp\Exception\RequestException;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
/**
* Demonstrates the Queue API.
*
* getData() loads external data and creates one queue item per entry in the
* "exqueue_import" queue. deleteTheQueue() empties that queue. On cron run the
* ExQueue01 queue worker turns each item into a page node.
*/
final class ExQueueController extends ControllerBase {
use AutowireTrait;
public function __construct(
#[Autowire(service: 'queue')]
protected readonly QueueFactory $queueFactory,
#[Autowire(service: 'http_client')]
protected readonly ClientInterface $httpClient,
) {}
/**
* Deletes the "exqueue_import" queue and all of its items.
*/
public function deleteTheQueue(): array {
$this->queueFactory->get('exqueue_import')->deleteQueue();
return ['#markup' => $this->t('The queue "exqueue_import" has been deleted.')];
}
/**
* Loads external data and pushes one queue item per entry.
*/
public function getData(): array {
// 1. Get the data as an array of objects (swap in getFakeData() locally).
$data = $this->getDataFromRss();
if (!$data) {
return ['#markup' => $this->t('No data found.')];
}
// 2. Get the queue and count its items before we add anything.
$queue = $this->queueFactory->get('exqueue_import');
$totalBefore = $queue->numberOfItems();
// 3. Create one queue item per entry.
foreach ($data as $element) {
$queue->createItem($element);
}
$totalAfter = $queue->numberOfItems();
// 4. Show what is now in the queue.
$list = $this->getItemList($queue);
return [
'#type' => 'table',
'#caption' => $this->t('The queue had @before item(s). We added @count. It now holds @after.', [
'@before' => $totalBefore,
'@count' => count($data),
'@after' => $totalAfter,
]),
'#header' => [$this->t('Title'), $this->t('ID')],
'#rows' => $list,
'#empty' => $this->t('No items.'),
'#sticky' => TRUE,
];
}
/**
* Builds a fake data set, handy for local testing without network access.
*/
protected function getFakeData(): array {
$content = [];
for ($i = 1; $i <= 10; $i++) {
$item = new \stdClass();
$item->title = 'Title ' . $i;
$item->body = 'Body ' . $i;
$content[] = $item;
}
return $content;
}
/**
* Fetches the Drupal Planet RSS feed and returns an array of item objects.
*/
protected function getDataFromRss(): array {
$uri = 'https://www.drupal.org/planet/rss.xml';
try {
$response = $this->httpClient->get($uri, ['headers' => ['Accept' => 'text/plain']]);
$body = (string) $response->getBody();
}
catch (RequestException) {
return [];
}
if ($body === '') {
return [];
}
$xml = simplexml_load_string($body);
if ($xml === FALSE) {
return [];
}
$content = [];
foreach ($xml->children()->children() as $child) {
if (!empty($child->title)) {
$item = new \stdClass();
$item->title = (string) $child->title;
$item->body = (string) $child->description;
$content[] = $item;
}
}
return $content;
}
/**
* Claims every item to display it, then releases the claims.
*/
protected function getItemList($queue): array {
$rows = [];
$claimed = [];
// claimItem() also leases (locks) the item for one hour by default, so we
// must release each claim afterwards.
while ($item = $queue->claimItem()) {
$rows[] = [$item->data->title, $item->item_id];
$claimed[] = $item;
}
foreach ($claimed as $item) {
$queue->releaseItem($item);
}
return $rows;
}
}Nous injectons ici deux services : QueueFactory pour travailler avec la queue, et le http_client de Guzzle pour récupérer le flux RSS. Les messages sont affichés avec $this->messenger(), que ControllerBase nous fournit gratuitement.
À présent, voici la partie intéressante : tout se passe dans getData() :
$queue = $this->queueFactory->get('exqueue_import');
$totalBefore = $queue->numberOfItems();
foreach ($data as $element) {
$queue->createItem($element);
}Trois lignes font le vrai travail. $this->queueFactory->get('exqueue_import') renvoie le backend QueueInterface par défaut pour une queue portant ce nom, créée à la première utilisation. numberOfItems() nous indique combien d'éléments attendent, et createItem() dépose un élément dans la queue.
Pour afficher la queue, nous la relisons ensuite : nous réclamons chaque élément puis nous le relâchons aussitôt, afin qu'il reste disponible pour le véritable worker :
while ($item = $queue->claimItem()) {
$rows[] = [$item->data->title, $item->item_id];
$claimed[] = $item;
}
foreach ($claimed as $item) {
$queue->releaseItem($item);
}Visitez /exqueue01/getData et la queue se remplit. Ça n'a pas l'air très compliqué, non ?
Pour inspecter la queue en ligne de commande, Drush 13 fournit une boîte à outils dédiée : drush queue:list affiche chaque queue enregistrée et son nombre d'éléments.
2. Vider la queue avec un plugin QueueWorker
Jusqu'ici tout va bien, mais une queue que personne ne vide ne sert pas à grand-chose. Il nous faut maintenant un QueueWorker pour traiter chaque élément sur le cron. En Drupal 11, le plugin est déclaré avec l'attribut PHP #[QueueWorker] : les annotations ont été supprimées en Drupal 10, donc l'ancien doc-block @QueueWorker ne fonctionne plus.
<?php
declare(strict_types=1);
namespace Drupal\exqueue01\Plugin\QueueWorker;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\Attribute\QueueWorker;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Symfony\Component\DependencyInjection\ContainerInterface;
/**
* Turns each queued RSS item into a page node.
*/
#[QueueWorker(
id: 'exqueue_import',
title: new TranslatableMarkup('Import Content From RSS'),
cron: ['time' => 5],
)]
final class ExQueue01 extends QueueWorkerBase implements ContainerFactoryPluginInterface {
public function __construct(
array $configuration,
$plugin_id,
$plugin_definition,
protected readonly EntityTypeManagerInterface $entityTypeManager,
protected readonly LoggerChannelFactoryInterface $loggerFactory,
) {
parent::__construct($configuration, $plugin_id, $plugin_definition);
}
public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition): self {
return new self(
$configuration,
$plugin_id,
$plugin_definition,
$container->get('entity_type.manager'),
$container->get('logger.factory'),
);
}
public function processItem($data): void {
$title = $data->title ?? NULL;
$body = $data->body ?? NULL;
// If the payload is unusable, throwing keeps the item in the queue; here we
// deliberately swallow the error so the useless item is dropped instead.
try {
if (!$title || !$body) {
throw new \InvalidArgumentException('Missing title or body.');
}
$node = $this->entityTypeManager->getStorage('node')->create([
'type' => 'page',
'title' => $title,
'body' => ['value' => $body, 'format' => 'basic_html'],
]);
$node->save();
$this->loggerFactory->get('exqueue01')->info('Created node @id from a queue item.', [
'@id' => $node->id(),
]);
}
catch (\Exception $e) {
$this->loggerFactory->get('exqueue01')->warning('Skipped a queue item: @error', [
'@error' => $e->getMessage(),
]);
}
}
}L'attribut relie le plugin à la queue :
#[QueueWorker(
id: 'exqueue_import',
title: new TranslatableMarkup('Import Content From RSS'),
cron: ['time' => 5],
)]id est le nom machine de la queue que ce worker va vider. La clé cron demande à Drupal de lancer le worker sur le cron et lui alloue jusqu'à 5 secondes par exécution ; les éléments restants sont repris au cron suivant.
À propos des exceptions. Sur le cron, la méthode Cron::processQueues() du cœur réclame chaque élément et appelle votre processItem() dans un bloc try/catch. Ce que vous lancez change le résultat :
- laissez une exception remonter et l'élément est conservé dans la queue et journalisé : utile quand l'échec est passager ;
- lancez
\Drupal\Core\Queue\RequeueExceptionpour remettre immédiatement l'élément en queue ; - lancez
\Drupal\Core\Queue\SuspendQueueExceptionpour arrêter le traitement de toute la queue (par ex. une API distante est en panne) ; - lancez
\Drupal\Core\Queue\DelayedRequeueExceptionpour conserver le bail mais réessayer plus tard.
Dans notre worker nous récupérons l'exception nous-mêmes, si bien qu'un élément défectueux est tout simplement abandonné plutôt que laissé indéfiniment dans la queue. C'est un choix de conception : un élément sans titre ni corps ne nous sert à rien.
3. Lancer le traitement de la queue
Deux façons de la vider :
- Cron : puisque nous avons défini la clé
cron, le worker s'exécute automatiquement à chaque passage du cron (drush cron, ou le cron planifié). - Drush 13 : lancez-le à la demande avec
drush queue:run exqueue_import. Cela remplace l'ancienne commande Drupal Consoledrupal queue:run.
En résumé. Nous avons construit un contrôleur qui récupère le flux RSS de Drupal Planet, créé une queue avec $this->queueFactory->get('exqueue_import'), déposé un élément par entrée avec createItem(), et écrit un plugin QueueWorker de cron dont la méthode processItem() transforme chaque élément en node de type page. Modernisées pour Drupal 11, les seules différences structurelles avec l'original Drupal 8 sont l'attribut #[QueueWorker], l'injection de dépendances par autowiring et la CLI Drush 13.
Vous remplissez vos queues autrement, depuis un formulaire ou directement dans une implémentation de hook_cron() ? Dites-le-nous dans les commentaires.
Pour en savoir plus