Issue #3301657 by pwolanin, jludwig: Add an alternate QueueFactory service to...

# Queue Unique

Did you ever wanted a queue that only accepts unique items? This module provides a way of doing that. If you try to insert a duplicated item in the queue, the item is ignored.

Did you ever want a queue that only accepts unique items? This module provides a way of doing that. If you try to insert a duplicated item in the queue, the item is ignored.

```php

// $data can be anything.

}

```

## Usage

In order for your queue to use the Queue Unique you need to update your `settings.php` file:

## Basic Usage

In order for your queue to use Queue Unique with the default queue service

(the core QueueFactory) you need to update your `settings.php` file:

```php

$settings['queue_service_your_queue_name'] = 'queue_unique.database';

```

Otherwise, you need to specifically get this module's database queue factory service:

```php

$queue_name = 'your_queue_name';

$queue = \Drupal::service('queue_unique.database')->get($queue_name);

$queue->createItem($data);

```

## Advanced Usage

This module provides an alternative QueueFactory class. Replace the core QueueFactory

using, for example, a site services.yaml file with an entry like this:

```

services:

  queue:

    class: Drupal\queue_unique\QueueFactory

    arguments: ['@settings']

    calls:

      - [setContainer, ['@service_container']]

```

See sites/default/default.services.yml and `$settings['container_yamls']` in default.settings.php.

When the QueueFactory is replaced, you can get a unique queue based on a prefix on the queue

name of `queue_unique.`

For example:

```php

$queue_name = 'queue_unique/your_queue_name';

$queue = \Drupal::service('queue')->get($queue_name);

$queue->createItem($data);

```

The actual queue name in the database will be `your_queue_name`.

This is especially useful with a queue worker plugin, e.g. extending `\Drupal\Core\Queue\QueueWorkerBase`

If you name the plugin ID with this prefix it can be processed

on cron automatically and correctly pull items from the unique queue:

```php

namespace Drupal\mymodule\Plugin\QueueWorker;

use Drupal\Core\Queue\QueueWorkerBase;

/**

 * Queue worker to handle when something has changed.

 *

 * @QueueWorker(

 *   id = "queue_unique/mymodule_entity_update",

 *   title = @Translation("Handle entity create or update."),

 *   cron = {"time" = 20}

 * )

 */

class EntityUpdateQueueWorker extends QueueWorkerBase {

}

```

Add items to the `mymodule_entity_update` unique queue and they will be processed on

cron. See \Drupal\Core\Cron::processQueues().

For example:

```

$queue = \Drupal::service('queue')->get("queue_unique/mymodule_entity_update");

$queue->createItem($data);

```

or:

```php

$queue = \Drupal::service('queue_unique.database')->get('mymodule_entity_update');

$queue->createItem($data);

```

<?php

namespace Drupal\queue_unique;

use Drupal\Core\Queue\QueueFactory as CoreQueueFactory;

/**

 * Defines the queue factory supporting queue_unique.

 */

class QueueFactory extends CoreQueueFactory {

  /**

   * Constructs a new queue. A name prefix can be used to find a queue factory.

   *

   * @param string $name

   *   The name of the queue to work with. If the name has a prefix ending with

   *   a "/" and a service exists with that prefix plus "database", we use

   *   that service to provide the queue by default. For example, if $name is

   *   "queue_unique/mymodule_work" we default to service

   *   "queue_unique.database" and queue name "mymodule_work". Settings for

   *   the specific queue service name and queue_default are used in preference

   *   to this default service name.

   * @param bool $reliable

   *   (optional) TRUE if the ordering of items and guaranteeing every item

   *   executes at least once is important, FALSE if scalability is the main

   *   concern. Defaults to FALSE.

   *

   * @return \Drupal\Core\Queue\QueueInterface

   *   A queue implementation for the given name.

   */

  public function get($name, $reliable = FALSE) {

    if (!isset($this->queues[$name])) {

      $service_name = NULL;

      $queue_name = $name;

      // If it is a reliable queue, check the specific settings first. This is

      // the same as the core factory.

      if ($reliable) {

        $service_name = $this->settings->get('queue_reliable_service_' . $name);

      }

      // If no reliable queue was defined, check the service and global

      // settings, then fall back to the default service name.

      if (empty($service_name)) {

        $service_name = $this->settings->get('queue_service_' . $name);

      }

      if (empty($service_name)) {

        [$default_service_name, $queue_name] = $this->defaultServiceAndQueueName($name);

        $service_name = $this->settings->get('queue_default', $default_service_name);

      }

      $this->queues[$name] = $this->container->get($service_name)->get($queue_name);

    }

    return $this->queues[$name];

  }

  /**

   * Build the default service name and final queue name from the name.

   *

   * @param string $name

   *   The name of the queue to work with.

   *

   * @return array

   *   The default service name and the queue name.

   */

  public function defaultServiceAndQueueName(string $name): array {

    $default_service_name = 'queue.database';

    // Check for a specific prefix. For example, "queue_unique/mymodule_work".

    $slash_pos = strpos($name, '/');

    if ($slash_pos) {

      $prefix = substr($name, 0, $slash_pos);

      $test_name = "$prefix.database";

      if ($this->container->has($test_name)) {

        // If the name was "queue_unique/mymodule_work" we end with service

        // "queue_unique.database" and queue name "mymodule_work".

        $default_service_name = $test_name;

        $name = substr($name, $slash_pos + 1);

      }

    }

    return [$default_service_name, $name];

  }

}

<?php

namespace Drupal\Tests\queue_unique\Kernel;

use Drupal\Core\DependencyInjection\ContainerBuilder;

use Drupal\Core\Queue\DatabaseQueue;

use Drupal\Core\Queue\QueueInterface;

use Drupal\KernelTests\KernelTestBase;

use Drupal\queue_unique\QueueFactory;

use Drupal\queue_unique\UniqueDatabaseQueue;

use Symfony\Component\DependencyInjection\Reference;

/**

 * Unique queue factory kernel test.

 *

 * @group queue_unique

 */

class QueueFactoryTest extends KernelTestBase {

  /**

   * {@inheritdoc}

   */

  protected static $modules = ['queue_unique'];

  /**

   * {@inheritdoc}

   */

  public function register(ContainerBuilder $container) {

    parent::register($container);

    // Update Settings before it's used as a service argument.

    foreach (['queue_unique/reliable', 'something_else'] as $name) {

      $this->setSetting('queue_reliable_service_' . $name, 'queue.database');

    }

    foreach (['queue/hello', 'something_unique'] as $name) {

      $this->setSetting('queue_service_' . $name, 'queue_unique.database');

    }

    $this->container->setParameter('install_profile', 'testing');

    // Replace core service.

    $this->container->register('queue', QueueFactory::class)

      ->addArgument(new Reference('settings'))

      ->addMethodCall('setContainer', [new Reference('service_container')]);

    // Add an alias for queue_unique.database.

    $this->container->addAliases(['llama_monster.database' => 'queue_unique.database']);

  }

  /**

   * Test that queues are found based on name/prefix.

   */

  public function testQueueNameMapping() {

    $queue_service = $this->container->get('queue');

    self::assertClassOf(QueueFactory::class, $queue_service);

    $queue_instance_count = 0;

    /* @var \Drupal\Core\Queue\QueueInterface $queue */

    $queue = $queue_service->get('queue_unique/test');

    self::assertClassOf(UniqueDatabaseQueue::class, $queue);

    self::assertQueueName($queue, 'test');

    $queue_instance_count++;

    // Test the aliased service.

    /* @var \Drupal\Core\Queue\QueueInterface $queue */

    $queue = $queue_service->get('llama_monster/test');

    self::assertClassOf(UniqueDatabaseQueue::class, $queue);

    self::assertQueueName($queue, 'test');

    $queue_instance_count++;

    /* @var \Drupal\Core\Queue\QueueInterface $queue */

    $queue = $queue_service->get('test');

    self::assertInstanceOf(DatabaseQueue::class, $queue);

    self::assertQueueName($queue, 'test');

    $queue_instance_count++;

    // This maps to core queue.database.

    /* @var \Drupal\Core\Queue\QueueInterface $queue */

    $queue = $queue_service->get('queue/test');

    self::assertClassOf(DatabaseQueue::class, $queue);

    self::assertQueueName($queue, 'test');

    $queue_instance_count++;

    /* @var \Drupal\Core\Queue\QueueInterface $queue */

    $queue = $queue_service->get('random/test');

    self::assertClassOf(DatabaseQueue::class, $queue);

    self::assertQueueName($queue, 'random/test');

    $queue_instance_count++;

    // Test Settings used in preference to the name service mapping.

    // See self::register() above.

    foreach (['queue_unique/reliable', 'something_else'] as $name) {

      /* @var \Drupal\Core\Queue\QueueInterface $queue */

      $queue = $queue_service->get($name, TRUE);

      self::assertClassOf(DatabaseQueue::class, $queue);

      self::assertQueueName($queue, $name);

      $queue_instance_count++;

    }

    foreach (['queue/hello', 'something_unique'] as $name) {

      $queue = $queue_service->get($name);

      self::assertClassOf(UniqueDatabaseQueue::class, $queue);

      self::assertQueueName($queue, $name);

      $queue_instance_count++;

    }

    // Every name should have generated a new queue instance.

    $reflected_queues = (new \ReflectionObject($queue_service))->getProperty('queues');

    $reflected_queues->setAccessible(TRUE);

    $queue_instances = $reflected_queues->getValue($queue_service);

    self::assertCount($queue_instance_count, $queue_instances);

    // Getting the same names again should not change the number of instances.

    foreach (['queue_unique/test', 'test', 'queue/test', 'random/test'] as $name) {

      $queue_service->get($name);

    }

    $queue_instances = $reflected_queues->getValue($queue_service);

    self::assertCount($queue_instance_count, $queue_instances);

    // Appending any string to the name will generate new instances.

    foreach (['queue_unique/test', 'test', 'queue/test', 'random/test'] as $name) {

      $queue_service->get($name . '2');

      $queue_instance_count++;

    }

    $queue_instances = $reflected_queues->getValue($queue_service);

    self::assertCount($queue_instance_count, $queue_instances);

  }

  /**

   * Check for exact match of class name to object.

   *

   * Note that self::assertInstanceOf() can give an unexpected pass since

   * UniqueDatabaseQueue is a subclass of DatabaseQueue.

   *

   * @param $class

   *   The class name.

   * @param $object

   *   The object

   */

  protected static function assertClassOf($class, $object) {

    self::assertSame($class, get_class($object));

  }

  /**

   * Verify that the name property of the queue is the expected value.

   *

   * @param \Drupal\Core\Queue\QueueInterface $queue

   *   A queue.

   * @param string $expected

   *   The expected name of the queue used in the database.

   *

   * @throws \ReflectionException

   */

  protected static function assertQueueName(QueueInterface $queue, string $expected): void {

    $reflected_name = (new \ReflectionObject($queue))->getProperty('name');

    $reflected_name->setAccessible(TRUE);

    self::assertSame($expected, $reflected_name->getValue($queue));

  }

  /**

   * Data provider with queue names to test.

   *

   * @return array

   *  A name.

   */

  public function queueNamesProvider(): array {

    return [

      ['queue_unique/fellow', 'queue_unique.database', 'fellow'],

      ['queue/hello', 'queue.database', 'hello'],

      ['llama_monster/test', 'llama_monster.database', 'test'],

      ['yellow_bird/bye', 'queue.database', 'yellow_bird/bye'],

    ];

  }

  /**

   * Test QueueService::defaultServiceAndQueueName().

   *

   * @dataProvider queueNamesProvider

   */

  public function testQueueNameAndDefaultService($input_name, $expected_service, $expected_name) {

    $queue_service = $this->container->get('queue');

    self::assertClassOf(QueueFactory::class, $queue_service);

    [$default_service_name, $name] = $queue_service->defaultServiceAndQueueName($input_name);

    self::assertEquals($expected_service, $default_service_name);

    self::assertEquals($expected_name, $name);

  }

}