无法使用 setTimeout 解决承诺

Question

我正在编写一个函数来使用 Node.js 写入 AWS Kinesis firehose。写入 firehose 流的 AWS 函数调用根据操作结果返回“错误”或“数据”。如果出现错误，对于特定的错误代码，我需要使用exponentialBackoff 重试相同的请求。我正在使用 setTimeOut 为每次后续重试触发相同的方法，但时间不同，但看起来每当重试时，我都没有正确解决，我的测试失败并抱怨“错误：超过 5000 毫秒的超时。对于异步测试和钩子，确保调用了“done()”；如果返回一个 Promise，确保它解析。"

async function batchWrite(records,firehose,retry = 0){
    var readingObjects = getReadings(records);
    var params = {
        DeliveryStreamName: process.env.KINESIS_FIREHOSE_STREAM_DELIVERY,
        Records: readingObjects
    };

    return await new Promise(function(resolve,reject){
        firehose.putRecordBatch(params,function(error,data){
            if(error){
                if(error.code == 'ServiceUnavailableException' && retry < retries.length){
                    console.log('retryCount=',retry);
                    setTimeout(batchWrite,retries[retry],records,firehose,retry+1);
                    console.log('setTimeout',retry);
                }
                else{
                    // console.log('Error',error);
                    console.log('resolving');
                    resolve(error);
                }
            }
            else{
                if(data.FailedPutCount > 0){
                    //colect the RequestResponses which are not processed.
                    //index of those records is same as in request
                    //process those faied records again
                    console.log('Error',data);
                }
                resolve(data);
            }
        });
    });
}

对于上面的函数，在测试用例中，我通过返回具有 ServiceUnavailableException 的响应来模拟错误情况，以便它重试，但在所有重试之后，测试超时。有什么我需要更改才能使其正常工作的吗？

Answer 1

当您调用setTimeout() 进行重试时，您创建和返回的原始承诺永远不会得到解决。因此，调用者永远不会看到已解决的承诺。您需要将新的承诺链接到前一个承诺。而且，你不能直接用setTimeout() 做到这一点，但你可以做到这一点，我将它包装在它自己的承诺中，然后将其链接到原始承诺。此外，如果您将 putRecordBatch 承诺为最低级别并使用承诺而不是承诺和回调的混合来完成所有控制流，则代码更易于管理。

你可以这样做：

// utility function to return a promise that is resolved after a setTimeout()
function delay(t, v) {
    return new Promise(resolve => {
        setTimeout(resolve, t, v);
    });
}

const promisify = require('util').promisify;

function batchWrite(records,firehose,retry = 0){
    var readingObjects = getReadings(records);
    var params = {
        DeliveryStreamName: process.env.KINESIS_FIREHOSE_STREAM_DELIVERY,
        Records: readingObjects
    };

    // make a promisified version of firehose.putRecordBatch
    firehose.putRecordBatchP = promisify(firehose.putRecordBatch);

    return firehose.putRecordBatchP(params).then(data => {
        if(data.FailedPutCount > 0){
            //collect the RequestResponses which are not processed.
            //index of those records is same as in request
            //process those failed records again
            console.log('Error',data);
        }
        return data;

    }).catch(error => {
        if(error.code == 'ServiceUnavailableException' && retry < retries.length){
            console.log('retryCount=',retry);

            // CHAIN the promise delay and retry here
            return delay(retries[retry]).then(() => {
                console.log('retry after delay #',retry);
                return batchWrite(records,firehose,retry+1); 
            });
        } else {
            // REJECT promise here upon error
            console.log('Error, rejecting promise');
            throw error;
        }

    });
}

关于你的代码的几件事我不明白：

1. 当您收到来自putRecordBatch() 的不可重试错误时，您似乎应该拒绝该承诺，以便调用者知道操作失败。为此我更改了代码。

2. 我不清楚if(data.FailedPutCount > 0){...} 子句的用途或为什么你所做的只是console.log()。