ES7承诺并等待在后台永远循环的异步功能

Question

这可能是一个特例：

我想从队列 (AWS SQS( 中读取，这是通过进行调用来完成的，该调用等待几秒钟的消息，然后解析 - 只要您想处理该队列，就会在循环中一次又一次地调用(它每次都会检查一个标志(。

这意味着我有一个consume函数，只要应用程序处于活动状态或队列未标记，它就会运行。

而且我还有一个用于订阅队列的subscribe函数 - 一旦知道消费者能够连接到队列，它就会立即解决。即使此函数调用保持运行并且在队列取消标记之前不会返回的使用者。

它给了我一些挑战 - 你有什么关于如何使用现代 JS 和 async/await 承诺解决这个问题的提示吗？我记住这段代码是在 React Web 应用程序中运行的，而不是在 node.js 中运行的。

我基本上只希望await subscribe(QUEUE)调用(来自 GUI(在确定它可以从该队列中读取后立即解决。但如果它不能，我希望它抛出一个错误，该错误传播到订阅调用的来源 - 这意味着我必须await consume(QUEUE)，对吧？

更新： 添加了一些未经测试的代码草稿(如果我没有做正确的方法，我不想花更多时间让它工作( - 我考虑过向消费函数发送成功和失败回调，这样它就可以在从队列中获得第一个有效(但可能是空的(响应后立即报告成功， 这使得它将队列 URL 存储为订阅 - 如果队列轮询失败，则取消订阅。

由于我正在设置多个队列使用者，因此它们不应该阻止任何东西，而只是在后台工作

let subscribedQueueURLs = []
async function consumeQueue(
url: QueueURL,
success: () => mixed,
failure: (error: Error) => mixed
) {
const sqs = new AWS.SQS()
const params = {
QueueUrl: url,
WaitTimeSeconds: 20,
}
try {
do {
// eslint-disable-next-line no-await-in-loop
const receivedData = await sqs.receiveMessage(params).promise()
if (!subscribedQueueURLs.includes(url)) {
success()
}
// eslint-disable-next-line no-restricted-syntax
for (const message of receivedData.Messages) {
console.log({ message })
// eslint-disable-next-line no-await-in-loop
eventHandler && (await eventHandler.message(message, url))
const deleteParams = {
QueueUrl: url,
ReceiptHandle: message.ReceiptHandle,
}
// eslint-disable-next-line no-await-in-loop
const deleteResult = await sqs.deleteMessage(deleteParams).promise()
console.log({ deleteResult })
}
} while (subscribedQueueURLs.includes(url))
} catch (error) {
failure(error)
}
}
export const subscribe = async (entityType: EntityType, entityId: EntityId) => {
const url = generateQueueURL(entityType, entityId)
consumeQueue(
url,
() => {
subscribedQueueURLs.push(url)
eventHandler && eventHandler.subscribe(url)
},
error => {
console.error(error)
unsubscribe(entityType, entityId)
}
)
}

Answer 1

我最终像这样解决了它 - 也许不是最优雅的解决方案......

let eventHandler: ?EventHandler
let awsOptions: ?AWSOptions
let subscribedQueueUrls = []
let sqs = null
let sns = null

export function setup(handler: EventHandler) {
eventHandler = handler
}
export async function login(
{ awsKey, awsSecret, awsRegion }: AWSCredentials,
autoReconnect: boolean
) {
const credentials = new AWS.Credentials(awsKey, awsSecret)
AWS.config.update({ region: awsRegion, credentials })
sqs = new AWS.SQS({ apiVersion: '2012-11-05' })
sns = new AWS.SNS({ apiVersion: '2010-03-31' })
const sts = new AWS.STS({ apiVersion: '2011-06-15' })
const { Account } = await sts.getCallerIdentity().promise()
awsOptions = { accountId: Account, region: awsRegion }
eventHandler && eventHandler.login({ awsRegion, awsKey, awsSecret }, autoReconnect)
}

async function handleQueueMessages(messages, queueUrl) {
if (!sqs) {
throw new Error(
'Attempt to subscribe before SQS client is ready (i.e. authenticated).'
)
}
// eslint-disable-next-line no-restricted-syntax
for (const message of messages) {
if (!eventHandler) {
return
}
// eslint-disable-next-line no-await-in-loop
await eventHandler.message({
content: message,
queueUrl,
timestamp: new Date().toISOString(),
})
const deleteParams = {
QueueUrl: queueUrl,
ReceiptHandle: message.ReceiptHandle,
}
// eslint-disable-next-line no-await-in-loop
await sqs.deleteMessage(deleteParams).promise()
}
}

export async function subscribe(queueUrl: QueueUrl) {
if (!sqs) {
throw new Error(
'Attempt to subscribe before SQS client is ready (i.e. authenticated).'
)
}
const initialParams = {
QueueUrl: queueUrl,
WaitTimeSeconds: 0,
MessageAttributeNames: ['All'],
AttributeNames: ['All'],
}
const longPollParams = {
...initialParams,
WaitTimeSeconds: 20,
}
// Attempt to consume the queue, and handle any pending messages.
const firstResponse = await sqs.receiveMessage(initialParams).promise()
if (!subscribedQueueUrls.includes(queueUrl)) {
subscribedQueueUrls.push(queueUrl)
eventHandler && eventHandler.subscribe(queueUrl)
}
handleQueueMessages(firstResponse.Messages, queueUrl)
// Keep on polling the queue afterwards.
setImmediate(async () => {
if (!sqs) {
throw new Error(
'Attempt to subscribe before SQS client is ready (i.e. authenticated).'
)
}
try {
do {
// eslint-disable-next-line no-await-in-loop
const received = await sqs.receiveMessage(longPollParams).promise()
handleQueueMessages(received.Messages, queueUrl)
} while (sqs && subscribedQueueUrls.includes(queueUrl))
} catch (error) {
eventHandler && eventHandler.disconnect()
throw error
}
})
}