Azure Service Bus Queue: можно ли управлять / отменять запланированные сообщения?

Question

Если я планирую сообщение в будущем, используя что-то вроде этого:

d = datetime.utcnow() + timedelta(minutes=5)
task = {"some": "object"}

sbs.send_queue_message(
    qn,
    Message(
        task, 
        broker_properties={'ScheduledEnqueueTimeUtc': d}
    )
)

Тогда есть ли способ просмотреть / удалить запланированные сообщения? send_queue_message ничего не возвращает, а receive_queue_message по понятным причинам не возвращает предметы, которые должны быть поставлены в очередь позже - поэтому я не могу достать его, например, до delete_queue_message.

Похоже, что команда Azure знает об этом сценарии использования, поскольку в очередях хранилища есть что-то вроде этой функции: https://azure.microsoft.com/en-gb/blog/azure-storage-queues-new-feature-pop-receipt-on-add-message/

По сути, мне нужно иметь возможность запланировать сообщение, которое будет поставлено в очередь позже, но его можно отменить. В идеале я хотел бы также иметь возможность просматривать все будущие запланированные задачи, но было бы достаточно просто сохранить идентификатор, который можно использовать для последующего удаления сообщения из очереди.

Пользовательский интерфейс Azure также показывает количество активных / запланированных сообщений, что, по-видимому, предполагает, что должен быть какой-то способ просмотреть эти запланированные!

Будет ли для этого лучше использовать хранилище в очереди? Или у служебной шины есть какой-то подход, который может работать? ScheduledEnqueueTimeUtc кажется более гибким, чем тайм-аут видимости в хранилище очереди, поэтому было бы неплохо придерживаться его, если смогу.

Answer 1

Для полноты картины это можно сделать с помощью службы очереди хранения Python SDK:

from azure.storage.queue import QueueService
account_name = '<snip>'
account_key = '<snip>'
queue_service = QueueService(account_name=account_name, account_key=account_key)
a = queue_service.put_message('queue_name', u'Hello World (delete)', visibility_timeout=30)
print(a.id) # id
print(a.pop_receipt) # pop_receipt

Затем в другом экземпляре Python до истечения времени ожидания видимости:

queue_service.delete_message('queue_name', id, pop_receipt)

Answer 2

Вы можете использовать «Microsoft.ServiceBus.Messaging» и удалять сообщения по времени в очереди. Получите сообщения, отфильтруйте по ScheduledEnqueueTime и выполните очистку, когда сообщение было поставлено в очередь в указанное время.

Microsoft.ServiceBus.Messaging;

MessagingFactory messagingFactory = MessagingFactory.CreateFromConnectionString (connectionString);

var queueClient = messagingFactory.CreateQueueClient (resourceName, ReceiveMode.PeekLock);

var client = messagingFactory.CreateMessageReceiver (resourceName, ReceiveMode.PeekLock);

Сообщение BrokeredMessage = client.Receive ();

            if (message.EnqueuedTimeUtc < MessageEnqueuedDateTime)
            {
                message.Complete(); 
            }

Answer 3

Да, это возможно.

Не знаю, поддерживает ли его клиент NodeJS или нет, но с клиентом C # есть альтернатива ScheduledEnqueueTimeUtc подходу, который я описал здесь . Используя QueueClient.ScheduleMessageAsync(), вы можете отправить запланированное сообщение и получить SequenceNumber. Который затем можно использовать для отмены сообщения в любой момент времени, используя QueueClient.CancelScheduledMessageAsync(sequenceNumber).