12 replies

Hey team, whats up?

Hey team, whats up?

I have a use case for queues and want to know if my logic is correct:
- I have a producer that posts around 500 messages at once on the queue, using batches of 25 messages
- All batches are delayed incrementally until the last batch on increments of 30 seconds each batch (first batch for 0 seconds, the second for 30 seconds, the third for 60 seconds...)
I was doing this process previously with a D1 column for the queue and a scheduled workers that runs each minute, but wanted to try it on using queues
The consumer should receive the batches following the delays, right?
The problem I'm facing: All the batches get processed at once, not obeying the delay sent while creating the batches.
I need this delay on the batches due to Slack API limitation that I use to process these messages, avoiding receiving 429s.
Is there any problem on the CF side causing this problem, or the queue will process all messages at once even with the delays?

Here are some snippets so you can take a look at my code:

queue definition

queue definition

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

producer

producer

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

consumer

consumer

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

Hey team, whats up?

queue definition

queue definition

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

[[queues.producers]]
  queue = "queue"
  binding = "QUEUE"

[[queues.consumers]]
  queue = "queue"
  max_batch_size = 25
  max_batch_timeout = 30
  max_concurrency = 1

producer

producer

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

let delaySeconds = 0
while (slackUsers.length) {
  console.log(delaySeconds)
  await triggerMessageQueue.sendBatch(
    slackUsers.splice(0, 25).map(({ id, dmChannel }) => ({
      body: {
        slackUserId: id,
        slackUserDmChannel: dmChannel,
        awarenessMessageId
      }
    })),
    { delaySeconds: delaySeconds }
  )
  delaySeconds += 30
}

consumer

consumer

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

export const triggerMessageQueueConsumer = async (
  batch: MessageBatch<QueueMessage>,
  env: QueueEnv
) => {
  console.log(`${new Date().toISOString()} Executing batch...`)
  console.log(`Received ${batch.messages.length}`)
  for (const message of batch.messages) {
    console.log(message.timestamp, message.attempts)
    message.ack()
  }
}

Hey team, whats up?

Hey team, whats up?

Continue the conversation

Cloudflare Developers

Continue the conversation

Cloudflare Developers

Similar Threads