Skip to content

Commit 4e6c0d2

Browse files
dreamorosidreamorosi
authored
feat(batch): add async processor (#1616)
* feat(batch): add async processor * tests: improved unit tests * chore: removed docstring + edited test handler
1 parent 49bf172 commit 4e6c0d2

19 files changed

+819
-461
lines changed

‎packages/batch/package.json

+3-4
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@aws-lambda-powertools/batch",
3-
"version": "1.10.0",
3+
"version": "1.11.1",
44
"description": "The batch processing package for the Powertools for AWS Lambda (TypeScript) library.",
55
"author": {
66
"name": "Amazon Web Services",
@@ -22,8 +22,7 @@
2222
"prepack": "node ../../.github/scripts/release_patch_package_json.js ."
2323
},
2424
"lint-staged": {
25-
"*.ts": "npm run lint-fix",
26-
"*.js": "npm run lint-fix"
25+
"*.{js,ts}": "npm run lint-fix"
2726
},
2827
"homepage": "https://github.com/aws-powertools/powertools-lambda-typescript/tree/main/packages/batch#readme",
2928
"license": "MIT-0",
@@ -50,4 +49,4 @@
5049
"nodejs"
5150
],
5251
"devDependencies": {}
53-
}
52+
}

‎packages/batch/src/AsyncBatchProcessor.ts

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
import { BasePartialBatchProcessor } from './BasePartialBatchProcessor';
2+
import type { BaseRecord, FailureResponse, SuccessResponse } from './types';
3+
4+
/**
5+
* Process native partial responses from SQS, Kinesis Data Streams, and DynamoDB
6+
*/
7+
class AsyncBatchProcessor extends BasePartialBatchProcessor {
8+
public async asyncProcessRecord(
9+
record: BaseRecord
10+
): Promise<SuccessResponse | FailureResponse> {
11+
try {
12+
const data = this.toBatchType(record, this.eventType);
13+
const result = await this.handler(data, this.options);
14+
15+
return this.successHandler(record, result);
16+
} catch (error) {
17+
return this.failureHandler(record, error as Error);
18+
}
19+
}
20+
21+
/**
22+
* Process a record with instance's handler
23+
* @param record Batch record to be processed
24+
* @returns response of success or failure
25+
*/
26+
public processRecord(_record: BaseRecord): SuccessResponse | FailureResponse {
27+
throw new Error('Not implemented. Use asyncProcess() instead.');
28+
}
29+
}
30+
31+
export { AsyncBatchProcessor };

‎packages/batch/src/BasePartialBatchProcessor.ts

+15-19
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,20 @@
1-
/**
2-
* Process batch and partially report failed items
3-
*/
4-
import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
5-
import {
6-
BasePartialProcessor,
7-
BatchProcessingError,
8-
DATA_CLASS_MAPPING,
9-
DEFAULT_RESPONSE,
1+
import type {
2+
DynamoDBRecord,
3+
KinesisStreamRecord,
4+
SQSRecord,
5+
} from 'aws-lambda';
6+
import { BasePartialProcessor } from './BasePartialProcessor';
7+
import { DATA_CLASS_MAPPING, DEFAULT_RESPONSE, EventType } from './constants';
8+
import { BatchProcessingError } from './errors';
9+
import type {
1010
EventSourceDataClassTypes,
11-
EventType,
12-
PartialItemFailures,
1311
PartialItemFailureResponse,
14-
} from '.';
12+
PartialItemFailures,
13+
} from './types';
1514

15+
/**
16+
* Process batch and partially report failed items
17+
*/
1618
abstract class BasePartialBatchProcessor extends BasePartialProcessor {
1719
public COLLECTOR_MAPPING;
1820

@@ -124,13 +126,7 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
124126
* @returns true if any records resulted in exception
125127
*/
126128
public hasMessagesToReport(): boolean {
127-
if (this.failureMessages.length != 0) {
128-
return true;
129-
}
130-
131-
// console.debug('All ' + this.successMessages.length + ' records successfully processed');
132-
133-
return false;
129+
return this.failureMessages.length != 0;
134130
}
135131

136132
/**

‎packages/batch/src/BasePartialProcessor.ts

+49-9
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
1-
/**
2-
* Abstract class for batch processors
3-
*/
4-
import {
1+
import type {
52
BaseRecord,
63
BatchProcessingOptions,
74
EventSourceDataClassTypes,
85
FailureResponse,
96
ResultType,
107
SuccessResponse,
11-
} from '.';
8+
} from './types';
129

10+
/**
11+
* Abstract class for batch processors.
12+
*/
1313
abstract class BasePartialProcessor {
1414
public exceptions: Error[];
1515

@@ -34,6 +34,40 @@ abstract class BasePartialProcessor {
3434
this.handler = new Function();
3535
}
3636

37+
/**
38+
* Call instance's handler for each record
39+
* @returns List of processed records
40+
*/
41+
public async asyncProcess(): Promise<(SuccessResponse | FailureResponse)[]> {
42+
/**
43+
* If this is an sync processor, user should have called process instead,
44+
* so we call the method early to throw the error early thus failing fast.
45+
*/
46+
if (this.constructor.name === 'BatchProcessor') {
47+
await this.asyncProcessRecord(this.records[0]);
48+
}
49+
this.prepare();
50+
51+
const processingPromises: Promise<SuccessResponse | FailureResponse>[] =
52+
this.records.map((record) => this.asyncProcessRecord(record));
53+
54+
const processedRecords: (SuccessResponse | FailureResponse)[] =
55+
await Promise.all(processingPromises);
56+
57+
this.clean();
58+
59+
return processedRecords;
60+
}
61+
62+
/**
63+
* Process a record with an asyncronous handler
64+
*
65+
* @param record Record to be processed
66+
*/
67+
public abstract asyncProcessRecord(
68+
record: BaseRecord
69+
): Promise<SuccessResponse | FailureResponse>;
70+
3771
/**
3872
* Clean class instance after processing
3973
*/
@@ -50,7 +84,6 @@ abstract class BasePartialProcessor {
5084
exception: Error
5185
): FailureResponse {
5286
const entry: FailureResponse = ['fail', exception.message, record];
53-
// console.debug('Record processing exception: ' + exception.message);
5487
this.exceptions.push(exception);
5588
this.failureMessages.push(record);
5689

@@ -66,12 +99,19 @@ abstract class BasePartialProcessor {
6699
* Call instance's handler for each record
67100
* @returns List of processed records
68101
*/
69-
public async process(): Promise<(SuccessResponse | FailureResponse)[]> {
102+
public process(): (SuccessResponse | FailureResponse)[] {
103+
/**
104+
* If this is an async processor, user should have called processAsync instead,
105+
* so we call the method early to throw the error early thus failing fast.
106+
*/
107+
if (this.constructor.name === 'AsyncBatchProcessor') {
108+
this.processRecord(this.records[0]);
109+
}
70110
this.prepare();
71111

72112
const processedRecords: (SuccessResponse | FailureResponse)[] = [];
73113
for (const record of this.records) {
74-
processedRecords.push(await this.processRecord(record));
114+
processedRecords.push(this.processRecord(record));
75115
}
76116

77117
this.clean();
@@ -85,7 +125,7 @@ abstract class BasePartialProcessor {
85125
*/
86126
public abstract processRecord(
87127
record: BaseRecord
88-
): Promise<SuccessResponse | FailureResponse>;
128+
): SuccessResponse | FailureResponse;
89129

90130
/**
91131
* Set class instance attributes before execution

‎packages/batch/src/BatchProcessor.ts

+13-14
Original file line numberDiff line numberDiff line change
@@ -1,30 +1,29 @@
1+
import { BasePartialBatchProcessor } from './BasePartialBatchProcessor';
2+
import type { BaseRecord, FailureResponse, SuccessResponse } from './types';
3+
14
/**
25
* Process native partial responses from SQS, Kinesis Data Streams, and DynamoDB
36
*/
4-
import {
5-
BasePartialBatchProcessor,
6-
BaseRecord,
7-
FailureResponse,
8-
SuccessResponse,
9-
} from '.';
10-
117
class BatchProcessor extends BasePartialBatchProcessor {
8+
public async asyncProcessRecord(
9+
_record: BaseRecord
10+
): Promise<SuccessResponse | FailureResponse> {
11+
throw new Error('Not implemented. Use process() instead.');
12+
}
13+
1214
/**
1315
* Process a record with instance's handler
1416
* @param record Batch record to be processed
1517
* @returns response of success or failure
1618
*/
17-
public async processRecord(
18-
record: BaseRecord
19-
): Promise<SuccessResponse | FailureResponse> {
19+
public processRecord(record: BaseRecord): SuccessResponse | FailureResponse {
2020
try {
2121
const data = this.toBatchType(record, this.eventType);
22-
23-
const result = await this.handler(data, this.options);
22+
const result = this.handler(data, this.options);
2423

2524
return this.successHandler(record, result);
26-
} catch (e) {
27-
return this.failureHandler(record, e as Error);
25+
} catch (error) {
26+
return this.failureHandler(record, error as Error);
2827
}
2928
}
3029
}

‎packages/batch/src/SqsFifoPartialProcessor.ts

+5-4
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1-
import { BatchProcessor, EventType, FailureResponse, SuccessResponse } from '.';
1+
import { BatchProcessor } from './BatchProcessor';
2+
import { EventType } from './constants';
3+
import type { FailureResponse, SuccessResponse } from './types';
24

35
/**
46
* Process native partial responses from SQS FIFO queues
@@ -14,9 +16,8 @@ class SqsFifoPartialProcessor extends BatchProcessor {
1416
* Call instance's handler for each record.
1517
* When the first failed message is detected, the process is short-circuited
1618
* And the remaining messages are reported as failed items
17-
* TODO: change to synchronous execution if possible
1819
*/
19-
public async process(): Promise<(SuccessResponse | FailureResponse)[]> {
20+
public process(): (SuccessResponse | FailureResponse)[] {
2021
this.prepare();
2122

2223
const processedRecords: (SuccessResponse | FailureResponse)[] = [];
@@ -28,7 +29,7 @@ class SqsFifoPartialProcessor extends BatchProcessor {
2829
return this.shortCircuitProcessing(currentIndex, processedRecords);
2930
}
3031

31-
processedRecords.push(await this.processRecord(record));
32+
processedRecords.push(this.processRecord(record));
3233
currentIndex++;
3334
}
3435

‎packages/batch/src/asyncProcessPartialResponse.ts

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
import { BasePartialBatchProcessor } from './BasePartialBatchProcessor';
2+
import { EventType } from './constants';
3+
import type {
4+
BaseRecord,
5+
BatchProcessingOptions,
6+
PartialItemFailureResponse,
7+
} from './types';
8+
9+
/**
10+
* Higher level function to handle batch event processing
11+
* @param event Lambda's original event
12+
* @param recordHandler Callable function to process each record from the batch
13+
* @param processor Batch processor to handle partial failure cases
14+
* @returns Lambda Partial Batch Response
15+
*/
16+
const asyncProcessPartialResponse = async (
17+
event: { Records: BaseRecord[] },
18+
recordHandler: CallableFunction,
19+
processor: BasePartialBatchProcessor,
20+
options?: BatchProcessingOptions
21+
): Promise<PartialItemFailureResponse> => {
22+
if (!event.Records) {
23+
const eventTypes: string = Object.values(EventType).toString();
24+
throw new Error(
25+
'Failed to convert event to record batch for processing.\nPlease ensure batch event is a valid ' +
26+
eventTypes +
27+
' event.'
28+
);
29+
}
30+
31+
processor.register(event.Records, recordHandler, options);
32+
33+
await processor.asyncProcess();
34+
35+
return processor.response();
36+
};
37+
38+
export { asyncProcessPartialResponse };

‎packages/batch/src/constants.ts

+4-4
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
/**
2-
* Constants for batch processor classes
3-
*/
41
import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
5-
import type { PartialItemFailureResponse, EventSourceDataClassTypes } from '.';
2+
import type {
3+
PartialItemFailureResponse,
4+
EventSourceDataClassTypes,
5+
} from './types';
66

77
const EventType = {
88
SQS: 'SQS',

‎packages/batch/src/errors.ts

-4
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,3 @@
1-
/**
2-
* Batch processing exceptions
3-
*/
4-
51
/**
62
* Base error type for batch processing
73
* All errors thrown by major failures extend this base class

‎packages/batch/src/index.ts

+2
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,7 @@ export * from './types';
44
export * from './BasePartialProcessor';
55
export * from './BasePartialBatchProcessor';
66
export * from './BatchProcessor';
7+
export * from './AsyncBatchProcessor';
78
export * from './processPartialResponse';
9+
export * from './asyncProcessPartialResponse';
810
export * from './SqsFifoPartialProcessor';

‎packages/batch/src/processPartialResponse.ts

+7-7
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
1-
import {
2-
BasePartialBatchProcessor,
1+
import { BasePartialBatchProcessor } from './BasePartialBatchProcessor';
2+
import { EventType } from './constants';
3+
import type {
34
BaseRecord,
45
BatchProcessingOptions,
5-
EventType,
66
PartialItemFailureResponse,
7-
} from '.';
7+
} from './types';
88

99
/**
1010
* Higher level function to handle batch event processing
@@ -13,12 +13,12 @@ import {
1313
* @param processor Batch processor to handle partial failure cases
1414
* @returns Lambda Partial Batch Response
1515
*/
16-
const processPartialResponse = async (
16+
const processPartialResponse = (
1717
event: { Records: BaseRecord[] },
1818
recordHandler: CallableFunction,
1919
processor: BasePartialBatchProcessor,
2020
options?: BatchProcessingOptions
21-
): Promise<PartialItemFailureResponse> => {
21+
): PartialItemFailureResponse => {
2222
if (!event.Records) {
2323
const eventTypes: string = Object.values(EventType).toString();
2424
throw new Error(
@@ -30,7 +30,7 @@ const processPartialResponse = async (
3030

3131
processor.register(event.Records, recordHandler, options);
3232

33-
await processor.process();
33+
processor.process();
3434

3535
return processor.response();
3636
};

0 commit comments

Comments
 (0)