C
MCCCC
M
M
D
AAA
ACMCCCMC
SSMS
KIs there a way to use the existing worker to subscribe to 2 queues at the same time?
SS
USCMCC
QUEUE.send() within a consumer, that it doesn't spawn a new worker within the same "thread"CQueue sendBatch failed: Subrequest depth limit exceeded. This request recursed through Workers too many times. but the message that has invoked the consumer should not reach this limit at allCCCKUCCCC
max_concurrent is default autoKC1QUEUE.send()Queue sendBatch failed: Subrequest depth limit exceeded. This request recursed through Workers too many times.export const handleMessage = async ({
message,
env,
}) => {
const { items, nextPageToken } =
await api.listByStatus({
limit: 100,
nextPageToken: message.nextPageToken,
status: 'approved',
});
if (nextPageToken) {
await sendQueueMessage({
message: {
type: '<REDACTED>',
nextPageToken,
},
env,
});
}
const toBeProcessed = items
.filter((f) => !f.processed)
await sendBatchQueueMessage({
batch: toBeProcessed.map((transaction) => ({
message: {
transaction,
type: '<OTHERMESSAGE>',
},
})),
env,
});
};export const sendQueueMessage = ({ message, env }: SendQueueMessage) => {
return env.PROCESS_QUEUE.send(message, { contentType: 'json' });
};
export const sendBatchQueueMessage = ({
batch,
env,
}: SendBatchQueueMessage) => {
const ch = chunk(batch, 20);
return Promise.all(
ch.map((c) => {
return env.PROCESS_QUEUE.sendBatch(
c.map((body) => ({ body: body.message, contentType: 'json' })),
);
}),
);
};await api.listByStatus({
limit: 100,max_concurrentauto[[queues.consumers]]
queue = "1"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optional
[[queues.consumers]]
queue = "2"
max_batch_size = 100 # optional
max_batch_timeout = 30 # optionalexport default {
async queue(batch: MessageBatch<Error>, env: Environment): Promise<void> {
switch (batch.queue) {
case '1':
// handle queue 1
break;
case '2':
// handle queue 2
break;
}
},
};