Dynamic consumers

[ebeyrent]

I'm working on a project where the queues are dynamically generated along with some associated config. The problem has been figuring out how to spin up consumers in response to the creation of the queues.

Let's say that we have a message entity with a field for a queue configuration. On the message entity postSave() method, we look at the value of that queue configuration field, and create the queue if needed, and then add the message entity to that queue.

$queue_name = $this->get('message_queue')->getValue();
 /** @var \Drupal\Core\Queue\QueueFactory $queue_factory */
$queue_factory = \Drupal::service('queue');
$queue = $queue_factory->get($queue_name[0]['value']);
$queue->createQueue();
$item = [
  'message_id' => $this->id(),
];
$queue->createItem($item);

In this example, we create the entity and when it gets saved, a queue is created and the entity is added to that queue.

In order to process that queue, I need a consumer to be running. Ideally, when the queue is created, an event would be dispatched and some eventsubscriber would catch it and cause a consumer process to get created. It looks like this project relies on known queue names for managing the consumers. Have you given any thought to how the consumers might be created dynamically?

[robiningelbrecht]

There are 2 things that come in play here:

• Creating and queue'ing messages to dynamic queues
• Starting consumers (actual processes) when a new queue is created

The first thing can be easily done by introducing a "DynamicQueue" class and passing in the name of the queue to the constructor.

The second thing is much harder as you will need to keep track of the dynamically created Qs somewhere to make sure, if you deploy or the code changes, all consumers can be restarted and pick up the code changes. Also you would need to start a new consumer through PHP, which I do not recommend. I personally don't like managing (unix) processes through PHP.

My question is, why would you want to do this? What is the use-case you are trying to solve?

[ebeyrent]

The idea here is that you have an application that needs to send emails. Some emails should be sent immediately; some should be sent during business hours; some should be sent on a throttled basis, i.e. x per unit of time. These configurations are tied to an actual queue, where the machine name of the config is the name of the queue. Now imagine that you need to add another queue, that only sends messages on the weekends.

The problem is that there's a tight coupling between how Drupal processes these queues - you either need a cron-based queue worker where the annotation specifies the queue name, or you need a separate task, like a drush command that takes a queue name as an argument and processes the queue. In that case, you still need something to fetch all the available queue names to create instances of the workers.

The problem I'm trying to solve is exactly that - how to automatically spin up queue workers when a new queue is created. You can further imagine that a queue configuration, besides days of the week, might contain a property to specify how many queue workers to spin up for that particular queue. A high priority queue might want three queue workers running in parallel, but a low priority queue might only need one.

[robiningelbrecht]

@ebeyrent Can't this be solved with the "Delayed Queue" examples I provided? These get created dynamically and will delay messages for X seconds / minutes / hours.

For example, if you want to send emails during business hours, you can

if($currentlyDuringBusinessHours) { 
   // Consumer(s) will pick up immediately.
   $this->someNormalQueue->queue($notification) ;
   return;
}

// Delayed Q will postpone consuming.
$this->delayedQueueFactory->buildWithDelayForQueue($secondsToBusinessHours, $someQueue)->queue($notification);

?

Now imagine that you need to add another queue, that only sends messages on the weekends.

To my knowledge a Q never sends messages? The worker for that Q will send out messages or am I missing something? :)

[ebeyrent]

Right - it's the workers I'm mostly concerned with, spinning those up dynamically

[robiningelbrecht]

@ebeyrent If you really want to talk this through, I'm open to set up a google meet video call or zoom call... I have a feeling this is a complicated thing you are trying to achieve and it's hard to explain this in words :)?