Skip to content

Commit

Permalink
First submit all discovered data before waiting for results
Browse files Browse the repository at this point in the history
  • Loading branch information
Fried Hoeben committed Nov 25, 2024
1 parent 1e3a8d3 commit 6f7b442
Show file tree
Hide file tree
Showing 2 changed files with 81 additions and 34 deletions.
92 changes: 67 additions & 25 deletions lansweeper/aws/integration-lambda/lansweeper_integration.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,21 +70,17 @@ class LansweeperIntegration {
if (selectedInstallations.length > 1) {
console.log('Will process the following installations: %j', selectedInstallations.map(i => i.name));
}
for (const installation of selectedInstallations) {
const installationResult = await this.processSite(siteId,
networkedAssetsOnly,
generateLabels,
installation,
installationNames,
assetTypes);
result.uploadCounts[siteName] = result.uploadCounts[siteName] || {};
result.uploadCounts[siteName][installation.name] = installationResult.uploadCount;
if (installationResult.errors) {
result.errorCounts[siteName] = result.errorCounts[siteName] || {};
result.errorCounts[siteName][installation.name] = installationResult.errors.length;
result.errors[siteName] = result.errors[siteName] || {};
result.errors[siteName][installation.name] = installationResult.errors;
}
const resultDownloadFunctions =
await this.enqueueMutationsForInstallations(selectedInstallations,
siteId,
networkedAssetsOnly,
generateLabels,
installationNames,
assetTypes);
if (resultDownloadFunctions.size > 0) {
console.log('Assets from all installations uploaded');
await this.downloadAndReduceSiteResults(resultDownloadFunctions, result, siteName);
console.log('All results gathered');
}
}
}
Expand All @@ -109,17 +105,63 @@ class LansweeperIntegration {
return this.resultHelper.cleanupResult(result);
}

async processSite(siteId, networkedAssetsOnly, generateLabels, installation, installationNames, assetTypes) {
async enqueueMutationsForInstallations(selectedInstallations, siteId, networkedAssetsOnly, generateLabels, installationNames, assetTypes) {
const resultDownloadFunctions = new Map();
for (const installation of selectedInstallations) {
const resultDownloadFunction =
await this.enqueueMutationsForInstallation(siteId,
networkedAssetsOnly,
generateLabels,
installation,
installationNames,
assetTypes);
resultDownloadFunctions.set(installation, resultDownloadFunction);
}
return resultDownloadFunctions;
}

async downloadAndReduceSiteResults(resultDownloadFunctions, result, siteName) {
const downloadCalls =
Array.from(resultDownloadFunctions,
async ([installation, resultDownloadFunction]) => {
const installationResult = await resultDownloadFunction.call();
return await this.mergeInstallationResult(result, siteName, installation, installationResult);
});
console.log('Waiting for all asyncQuery results to become available');
await Promise.all(downloadCalls);
}

async mergeInstallationResult(result, siteName, installation, installationResult) {
result.uploadCounts[siteName] = result.uploadCounts[siteName] || {};
result.uploadCounts[siteName][installation.name] = installationResult.uploadCount;
if (installationResult.errors) {
result.errorCounts[siteName] = result.errorCounts[siteName] || {};
result.errorCounts[siteName][installation.name] = installationResult.errors.length;
result.errors[siteName] = result.errors[siteName] || {};
result.errors[siteName][installation.name] = installationResult.errors;
}
}

async enqueueMutationsForInstallation(siteId, networkedAssetsOnly, generateLabels, installation, installationNames, assetTypes) {
const siteName = await this.lansweeperClient.getSiteName(siteId);
const installationName = installation.name;
console.log(`processing site ${siteName}, installation ${installationName}. NetworkedAssetsOnly: ${networkedAssetsOnly}.${generateLabels ? ' Using asset name as label.' : ''}`);
const itemsHandler = async items => await this.sendAssetsTo4me(items, networkedAssetsOnly, generateLabels, installationName, installationNames);
const sendResults = await this.lansweeperClient.getAssetsPaged(siteId, this.assetSeenCutOffDate(), itemsHandler, networkedAssetsOnly, installation.id, assetTypes);
const jsonResults = await this.downloadResults(sendResults.map(r => r.mutationResult));
const overallResult = this.reduceResults(sendResults, jsonResults);
console.log(`processed site ${siteName}, installation ${installationName}. Upload count: ${overallResult.uploadCount}, error count: ${overallResult.errors ? overallResult.errors.length : 0}`);
const sendResults = await this.startInstallationSync(siteId, networkedAssetsOnly, generateLabels, installation, installationName, installationNames, assetTypes);
console.log(`Uploaded all assets for site ${siteName}, installation ${installationName}.`);

// function to download all results for this installation and merge to single result
return async () => {
const jsonResults = await this.downloadResults(installationName, sendResults.map(r => r.mutationResult));
const overallResult = this.reduceResults(sendResults, jsonResults);
console.log(`processed site ${siteName}, installation ${installationName}. Upload count: ${overallResult.uploadCount}, error count: ${overallResult.errors ? overallResult.errors.length : 0}`);

return overallResult;
return overallResult;
};
}

async startInstallationSync(siteId, networkedAssetsOnly, generateLabels, installation, installationName, installationNames, assetTypes) {
const itemsHandler = async items => await this.sendAssetsTo4me(items, networkedAssetsOnly, generateLabels, installationName, installationNames);
return await this.lansweeperClient.getAssetsPaged(siteId, this.assetSeenCutOffDate(), itemsHandler, networkedAssetsOnly, installation.id, assetTypes);
}

async sendAssetsTo4me(assets, networkedAssetsOnly = false, generateLabels = false, installation = null, installations = []) {
Expand Down Expand Up @@ -169,12 +211,12 @@ class LansweeperIntegration {
return new Date(new TimeHelper().getMsSinceEpoch() - LansweeperIntegration.LAST_SEEN_DAYS * 24 * 60 * 60 * 1000);
}

async downloadResults(mutationResultsToRetrieve) {
async downloadResults(installationName, mutationResultsToRetrieve) {
const jsonResults = new Map();
if (mutationResultsToRetrieve.length === 0) {
console.log('No asynchronous queries results to retrieve');
console.log(`No asynchronous queries results to retrieve for ${installationName}`);
} else {
console.log('Downloading all asynchronous query results');
console.log(`Downloading ${mutationResultsToRetrieve.length} asynchronous query results for ${installationName}`);
const jsonRetrievalCalls = mutationResultsToRetrieve
.filter(r => !!r)
.map(async r => jsonResults.set(r, await this.downloadResult(r)));
Expand Down
23 changes: 14 additions & 9 deletions lansweeper/aws/integration-lambda/tests/lansweeper_integration.test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ describe('processSite', () => {
}));
});

const processSite = async (integration, siteId, networkedAssetsOnly, generateLabels, installation, installationNames, assetTypes) => {
const resultDownloadFunction = await integration.enqueueMutationsForInstallation(siteId, networkedAssetsOnly, generateLabels, installation, installationNames, assetTypes);
return await resultDownloadFunction.call();
}

it('handles successful pages', async () => {
const siteId = 'abdv';

Expand Down Expand Up @@ -112,7 +117,7 @@ describe('processSite', () => {
'refresh token',
mockedJs4meHelper);

const result = await integration.processSite(siteId, true, true, installations[0], installations, null);
const result = await processSite(integration, siteId, true, true, installations[0], installations, null);
const uploadCount = graphQLResult.configurationItems.length * 2;
expect(result).toEqual({uploadCount: uploadCount});
});
Expand Down Expand Up @@ -147,8 +152,8 @@ describe('processSite', () => {
'refresh token',
null);

const result = await integration.processSite(siteId, false, false, installations[0], installations, null);
expect(result).toEqual({errors: ['Unable to upload assets to 4me.']});
const result = await processSite(integration, siteId, false, false, installations[0], installations, null);
expect(result).toEqual({errors: ['Unable to upload assets to 4me.'], uploadCount: 0});
expect(consoleErrorSpy).not.toHaveBeenCalled();
});

Expand Down Expand Up @@ -182,8 +187,8 @@ describe('processSite', () => {
'refresh token',
null);

const result = await integration.processSite(siteId, false, false, installations[0], installations, null);
expect(result).toEqual({errors: ['Unable to upload assets to 4me.']});
const result = await processSite(integration, siteId, false, false, installations[0], installations, null);
expect(result).toEqual({errors: ['Unable to upload assets to 4me.'], uploadCount: 0});
expect(consoleErrorSpy).toHaveBeenCalled();
});

Expand Down Expand Up @@ -261,7 +266,7 @@ describe('processSite', () => {
'refresh token',
mockedJs4meHelper);

const result = await integration.processSite(siteId, false, false, installations[0], installations, null);
const result = await processSite(integration, siteId, false, false, installations[0], installations, null);
const uploadCount = graphQLResult.configurationItems.length;
expect(result).toEqual({errors: ['Unable to upload'], uploadCount: uploadCount});
});
Expand Down Expand Up @@ -338,7 +343,7 @@ describe('processSite', () => {
'refresh token',
mockedJs4meHelper);

const result = await integration.processSite(siteId, undefined, false, installations[0], installations, null);
const result = await processSite(integration, siteId, undefined, false, installations[0], installations, null);
const uploadCount = graphQLResult.configurationItems.length * 2;
expect(result).toEqual({errors: ['unable to create ci1'], uploadCount: uploadCount});
});
Expand Down Expand Up @@ -402,7 +407,7 @@ describe('processSite', () => {
'refresh token',
mockedJs4meHelper);

await expect(integration.processSite(siteId, undefined, false, installations[0], installations, null))
await expect(processSite(integration, siteId, undefined, false, installations[0], installations, null))
.rejects
.toThrow(error);
});
Expand Down Expand Up @@ -798,7 +803,7 @@ describe('processSite', () => {
'refresh token',
mockedJs4meHelper);

const result = await integration.processSite(siteId, undefined, false, installations[0], installations, null);
const result = await processSite(integration, siteId, undefined, false, installations[0], installations, null);
const uploadCount = graphQLResult.configurationItems.length;
expect(result).toEqual({errors: ['Unable to query abc'], uploadCount: uploadCount});
});
Expand Down

0 comments on commit 6f7b442

Please sign in to comment.