data_pipelines-1.x-dev/tests/src/Traits/DatasetTestProcessingTrait.php

<?php

declare(strict_types=1);

namespace Drupal\Tests\data_pipelines\Traits;

use Drupal\data_pipelines\Destination\ProcessingOperation;

/**
 * Dataset Processing Trait.
 */
trait DatasetTestProcessingTrait {

  /**
   * Processes the data pipeline queue.
   *
   * @param string $queue_id
   *   The queue ID.
   * @param string $destination_id
   *   The destination ID.
   * @param array $operations
   *   The array of all the queued operations.
   */
  protected function processQueue(string $queue_id, string $destination_id, array $operations): void {
    /** @var \Drupal\Core\Queue\QueueWorkerInterface $queue_worker */
    $queue_worker = \Drupal::service('plugin.manager.queue_worker')->createInstance($queue_id);
    $process_queue = \Drupal::queue($queue_id);
    while ($process_queue->numberOfItems()) {
      $item = $process_queue->claimItem();
      $data = $item->data;
      assert($data instanceof ProcessingOperation);
      $this->assertEquals(current($operations), $data->getOperation());
      $this->assertEquals($destination_id, $data->getDestinationId());
      next($operations);
      $queue_worker->processItem($data);
      $process_queue->deleteItem($item);
    }
  }

  /**
   * Processes the data pipeline queue.
   *
   * @param string $queue_id
   *   The queue ID.
   * @param string $destination_id
   *   The destination ID.
   * @param array $operations
   *   The array of all the queued operations.
   */
  protected function processQueueWithWait(string $queue_id, string $destination_id, array $operations): void {
    $this->processQueue($queue_id, $destination_id, $operations);
    // Let ES/OS container do its thing.
    sleep(1);

  }

}