Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(cli): parallel asset publishing #19367

Merged
merged 3 commits into from
Mar 14, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/aws-cdk/lib/util/asset-publishing.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ export async function publishAssets(manifest: cdk_assets.AssetManifest, sdk: Sdk
aws: new PublishingAws(sdk, targetEnv),
progressListener: new PublishingProgressListener(),
throwOnError: false,
publishInParallel: true,
});
await publisher.publish();
if (publisher.hasFailures) {
Expand Down
75 changes: 49 additions & 26 deletions packages/cdk-assets/lib/publishing.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,13 @@ export interface AssetPublishingOptions {
* @default true
*/
readonly throwOnError?: boolean;

/**
* Whether to publish in parallel
*
* @default false
*/
readonly publishInParallel?: boolean;
}

/**
Expand Down Expand Up @@ -56,44 +63,33 @@ export class AssetPublishing implements IPublishProgress {
private readonly totalOperations: number;
private completedOperations: number = 0;
private aborted = false;
private readonly handlerHost: IHandlerHost;
private readonly publishInParallel: boolean;

constructor(private readonly manifest: AssetManifest, private readonly options: AssetPublishingOptions) {
this.assets = manifest.entries;
this.totalOperations = this.assets.length;
}
this.publishInParallel = options.publishInParallel ?? false;

/**
* Publish all assets from the manifest
*/
public async publish(): Promise<void> {
const self = this;

const handlerHost: IHandlerHost = {
this.handlerHost = {
aws: this.options.aws,
get aborted() { return self.aborted; },
emitMessage(t, m) { self.progressEvent(t, m); },
};
}

for (const asset of this.assets) {
if (this.aborted) { break; }
this.currentAsset = asset;

try {
if (this.progressEvent(EventType.START, `Publishing ${asset.id}`)) { break; }

const handler = makeAssetHandler(this.manifest, asset, handlerHost);
await handler.publish();

if (this.aborted) {
throw new Error('Aborted');
/**
* Publish all assets from the manifest
*/
public async publish(): Promise<void> {
if (this.publishInParallel) {
await Promise.all(this.assets.map(async (asset) => this.publishAsset(asset)));
} else {
for (const asset of this.assets) {
if (!await this.publishAsset(asset)) {
break;
}

this.completedOperations++;
if (this.progressEvent(EventType.SUCCESS, `Published ${asset.id}`)) { break; }
} catch (e) {
this.failures.push({ asset, error: e });
this.completedOperations++;
if (this.progressEvent(EventType.FAIL, e.message)) { break; }
}
}

Expand All @@ -102,6 +98,33 @@ export class AssetPublishing implements IPublishProgress {
}
}

/**
* Publish an asset.
* @param asset The asset to publish
* @returns false when publishing should stop
*/
private async publishAsset(asset: IManifestEntry) {
try {
if (this.progressEvent(EventType.START, `Publishing ${asset.id}`)) { return false; }

const handler = makeAssetHandler(this.manifest, asset, this.handlerHost);
await handler.publish();

if (this.aborted) {
throw new Error('Aborted');
}

this.completedOperations++;
if (this.progressEvent(EventType.SUCCESS, `Published ${asset.id}`)) { return false; }
} catch (e) {
this.failures.push({ asset, error: e });
this.completedOperations++;
if (this.progressEvent(EventType.FAIL, e.message)) { return false; }
}

return true;
}

public get percentComplete() {
if (this.totalOperations === 0) { return 100; }
return Math.floor((this.completedOperations / this.totalOperations) * 100);
Expand Down
13 changes: 13 additions & 0 deletions packages/cdk-assets/test/progress.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,19 @@ test('test listener', async () => {
expect(allMessages).toContain('theAsset:theDestination2');
});

test('test publishing in parallel', async () => {
const progressListener = new FakeListener();

const pub = new AssetPublishing(AssetManifest.fromPath('/simple/cdk.out'), { aws, progressListener, publishInParallel: true });
await pub.publish();

const allMessages = progressListener.messages.join('\n');

// Log mentions asset/destination ids
expect(allMessages).toContain('theAsset:theDestination1');
expect(allMessages).toContain('theAsset:theDestination2');
});

test('test abort', async () => {
const progressListener = new FakeListener(true);

Expand Down