https://stackoverflow.com/a/57540786/8784402
使用异步生成器
导入 S3
const { S3 } = require("aws-sdk");
const s3 = new S3();
创建一个生成器函数来检索所有文件列表
async function* listAllKeys(opts) {
opts = { ...opts };
do {
const data = await s3.listObjectsV2(opts).promise();
opts.ContinuationToken = data.NextContinuationToken;
yield data;
} while (opts.ContinuationToken);
}
const opts = {
Bucket: "bucket-xyz" /* required */,
// ContinuationToken: 'STRING_VALUE',
// Delimiter: 'STRING_VALUE',
// EncodingType: url,
// FetchOwner: true || false,
// MaxKeys: 'NUMBER_VALUE',
// Prefix: 'STRING_VALUE',
// RequestPayer: requester,
// StartAfter: 'STRING_VALUE'
};
使用生成器
async function main() {
// using for of await loop
for await (const data of listAllKeys(opts)) {
console.log(data.Contents);
}
}
main();
就是这样
或延迟加载
async function main() {
const keys = listAllKeys(opts);
console.log(await keys.next());
// {value: {…}, done: false}
console.log(await keys.next());
// {value: {…}, done: false}
console.log(await keys.next());
// {value: undefined, done: true}
}
main();
或者使用生成器制作 Observable 函数
const lister = (opts) => (o$) => {
let needMore = true;
const process = async () => {
for await (const data of listAllKeys(opts)) {
o$.next(data);
if (!needMore) break;
}
o$.complete();
};
process();
return () => (needMore = false);
};
在 RXJS 中使用这个 observable 函数
// Using Rxjs
const { Observable } = require("rxjs");
const { flatMap } = require("rxjs/operators");
function listAll() {
return Observable.create(lister(opts))
.pipe(flatMap((v) => v.Contents))
.subscribe(console.log);
}
listAll();
或者在 Nodejs EventEmitter 中使用这个 observable 函数
const EventEmitter = require("events");
const _eve = new EventEmitter();
async function onData(data) {
// will be called for each set of data
console.log(data);
}
async function onError(error) {
// will be called if any error
console.log(error);
}
async function onComplete() {
// will be called when data completely received
}
_eve.on("next", onData);
_eve.on("error", onError);
_eve.on("complete", onComplete);
const stop = lister(opts)({
next: (v) => _eve.emit("next", v),
error: (e) => _eve.emit("error", e),
complete: (v) => _eve.emit("complete", v),
});
使用 Typescript 和 AWS-SDK v3 + Deno
import {
paginateListObjectsV2,
S3Client,
S3ClientConfig,
} from "@aws-sdk/client-s3";
/* // For Deno
import {
paginateListObjectsV2,
S3Client,
S3ClientConfig,
} from "https://deno.land/x/aws_sdk@v3.14.0.0/client-s3/mod.ts"; */
const s3Config: S3ClientConfig = {
credentials: {
accessKeyId: "accessKeyId",
secretAccessKey: "secretAccessKey",
},
region: "us-east-1",
};
const client = new S3Client(s3Config);
const s3Opts = { Bucket: "bucket-xyz" };
async function getAllS3Files() {
const totalFiles = [];
for await (const data of paginateListObjectsV2({ client }, s3Opts)) {
totalFiles.push(...(data.Contents ?? []));
}
return totalFiles;
}