diff --git a/package-lock.json b/package-lock.json index fcd194f..64a1909 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "aws-lambda-stream", - "version": "1.1.14", + "version": "1.1.15", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "aws-lambda-stream", - "version": "1.1.14", + "version": "1.1.15", "license": "MIT", "dependencies": { "object-sizeof": "^2.6.0" diff --git a/package.json b/package.json index ebd915b..57b8ef0 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "aws-lambda-stream", - "version": "1.1.14", + "version": "1.1.15", "description": "Create stream processors with AWS Lambda functions.", "keywords": [ "aws", diff --git a/src/flavors/job.js b/src/flavors/job.js index f8daa87..d5a883e 100644 --- a/src/flavors/job.js +++ b/src/flavors/job.js @@ -1,8 +1,8 @@ -import _ from 'highland'; import { printStartPipeline, printEndPipeline, faulty, faultyAsyncStream, faultify, splitObject, encryptEvent, + compact, } from '../utils'; import { scanSplitDynamoDB, querySplitDynamoDB, queryAllDynamoDB, batchGetDynamoDB, @@ -132,35 +132,36 @@ export const toCursorUpdateRequest = (rule) => faulty((uow) => ({ })); export const flushCursor = (rule) => (s) => { - let lastUow; - - const cursorStream = () => _([lastUow]) - .map(toCursorUpdateRequest(rule)) - .through(updateDynamoDB({ - ...rule, - updateRequestField: 'cursorUpdateRequest', - updateResponseField: 'cursorUpdateResponse', - })); + const { + // By default group on a stringified version of the full key. If the key structure + // differs in a users particular implementation or they want to group by something + // else they can simply override this fn in their rule. + cursorKeyFn = (uow) => `pk:${uow.event.raw.new.pk}|sk:${uow.event.raw.new.sk}`, + } = rule; /* istanbul ignore else */ if (rule.toCursorUpdateRequest) { return s - .consume((err, x, push, next) => { - /* istanbul ignore if */ - if (err) { - push(err); - next(); - } else if (x === _.nil) { - if (lastUow) { - next(cursorStream()); - } else { - push(null, x); - } - } else { - lastUow = x; - push(null, x); - next(); - } + .through(compact({ + ...rule, + compact: { + group: (uow) => cursorKeyFn(uow), + }, + })) + .map(toCursorUpdateRequest(rule)) + .through(updateDynamoDB({ + ...rule, + updateRequestField: 'cursorUpdateRequest', + updateResponseField: 'cursorUpdateResponse', + })) + // Maintains backwards compatibility with how this used to manipulate the UOWs, + // duping the last uow. + .flatMap((uow) => { + const { batch, ...lastUow } = uow; + return [ + ...batch, + lastUow, + ]; }); } else { return s; diff --git a/test/unit/flavors/job.test.js b/test/unit/flavors/job.test.js index 073e221..66e5cc3 100644 --- a/test/unit/flavors/job.test.js +++ b/test/unit/flavors/job.test.js @@ -255,6 +255,320 @@ describe('flavors/job.js', () => { .done(done); }); + it('should propagate cursors across multiple job triggers in a single invocation', (done) => { + sinon.stub(DynamoDBConnector.prototype, 'queryPage') + .onCall(0) + .resolves({ + LastEvaluatedKey: { + pk: '4', + sk: 'thing', + }, + Items: [ + { + pk: '3', + sk: 'thing', + name: 'thing 3', + }, + { + pk: '4', + sk: 'thing', + name: 'thing 4', + }, + ], + }) + .onCall(1) + .resolves({ + LastEvaluatedKey: { + pk: '6', + sk: 'thing', + }, + Items: [ + { + pk: '5', + sk: 'thing', + name: 'thing 5', + }, + { + pk: '6', + sk: 'thing', + name: 'thing 6', + }, + ], + }); + + const events = toDynamodbRecords([ + { + timestamp: 1572832694, + keys: { + pk: 'job-1', + sk: 'job', + }, + newImage: { + pk: 'job-1', + sk: 'job', + discriminator: 'job', + cursor: { + pk: '2', + sk: 'thing', + }, + }, + oldImage: { + pk: 'job-1', + sk: 'job', + discriminator: 'job', + }, + }, { + timestamp: 1572832694, + keys: { + pk: 'job-2', + sk: 'job', + }, + newImage: { + pk: 'job-2', + sk: 'job', + discriminator: 'job', + cursor: { + pk: '4', + sk: 'thing', + }, + }, + oldImage: { + pk: 'job-2', + sk: 'job', + discriminator: 'job', + }, + }, + ]); + + initialize({ + ...initializeFrom(rules), + }, { ...defaultOptions, AES: false }) + .assemble(fromDynamodb(events), false) + .collect() + // .tap((collected) => console.log(JSON.stringify(collected, null, 2))) + .tap((collected) => { + // 1 per query split result and 1 per cursor. + expect(collected.length).to.equal(6); + + // First pk cursor processing + expect(collected[0].pipeline).to.equal('job1-continued'); + expect(collected[0].querySplitRequest).to.be.deep.equal({ + ExclusiveStartKey: { + pk: '2', + sk: 'thing', + }, + ExpressionAttributeNames: { + '#discriminator': 'discriminator', + }, + ExpressionAttributeValues: { + ':discriminator': 'thing', + }, + Limit: 2, + }); + expect(collected[0].querySplitResponse).to.be.deep.equal({ + LastEvaluatedKey: { + pk: '4', + sk: 'thing', + }, + Item: { + pk: '3', + sk: 'thing', + name: 'thing 3', + }, + }); + expect(collected[0].emit).to.deep.equal({ + type: 'xyz', + raw: { + pk: '3', + sk: 'thing', + name: 'thing 3', + }, + tags: { + account: 'undefined', + region: 'us-west-2', + stage: 'undefined', + source: 'undefined', + functionname: 'undefined', + pipeline: 'job1-continued', + skip: true, + }, + }); + expect(collected[0].cursorUpdateRequest).to.be.undefined; + expect(collected[1].querySplitRequest).to.be.deep.equal({ + ExclusiveStartKey: { + pk: '2', + sk: 'thing', + }, + ExpressionAttributeNames: { + '#discriminator': 'discriminator', + }, + ExpressionAttributeValues: { + ':discriminator': 'thing', + }, + Limit: 2, + }); + expect(collected[1].querySplitResponse).to.be.deep.equal({ + LastEvaluatedKey: { + pk: '4', + sk: 'thing', + }, + Item: { + pk: '4', + sk: 'thing', + name: 'thing 4', + }, + }); + expect(collected[1].emit).to.deep.equal({ + type: 'xyz', + raw: { + pk: '4', + sk: 'thing', + name: 'thing 4', + }, + tags: { + account: 'undefined', + region: 'us-west-2', + stage: 'undefined', + source: 'undefined', + functionname: 'undefined', + pipeline: 'job1-continued', + skip: true, + }, + }); + expect(collected[1].cursorUpdateRequest).to.be.undefined; + expect(collected[2].cursorUpdateRequest).to.deep.equal({ + Key: { + pk: 'job-1', + sk: 'job', + }, + ExpressionAttributeNames: { + '#cursor': 'cursor', + '#timestamp': 'timestamp', + }, + ExpressionAttributeValues: { + ':timestamp': 1572832694000, + ':cursor': { + pk: '4', + sk: 'thing', + }, + }, + UpdateExpression: 'SET #cursor = :cursor, #timestamp = :timestamp', + ReturnValues: 'ALL_NEW', + ConditionExpression: 'attribute_not_exists(#timestamp) OR #timestamp < :timestamp', + }); + // End first uow cursor processing + + // Second pk cursor processing + expect(collected[3].pipeline).to.equal('job1-continued'); + expect(collected[3].querySplitRequest).to.be.deep.equal({ + ExclusiveStartKey: { + pk: '4', + sk: 'thing', + }, + ExpressionAttributeNames: { + '#discriminator': 'discriminator', + }, + ExpressionAttributeValues: { + ':discriminator': 'thing', + }, + Limit: 2, + }); + expect(collected[3].querySplitResponse).to.be.deep.equal({ + LastEvaluatedKey: { + pk: '6', + sk: 'thing', + }, + Item: { + pk: '5', + sk: 'thing', + name: 'thing 5', + }, + }); + expect(collected[3].emit).to.deep.equal({ + type: 'xyz', + raw: { + pk: '5', + sk: 'thing', + name: 'thing 5', + }, + tags: { + account: 'undefined', + region: 'us-west-2', + stage: 'undefined', + source: 'undefined', + functionname: 'undefined', + pipeline: 'job1-continued', + skip: true, + }, + }); + expect(collected[3].cursorUpdateRequest).to.be.undefined; + expect(collected[4].querySplitRequest).to.be.deep.equal({ + ExclusiveStartKey: { + pk: '4', + sk: 'thing', + }, + ExpressionAttributeNames: { + '#discriminator': 'discriminator', + }, + ExpressionAttributeValues: { + ':discriminator': 'thing', + }, + Limit: 2, + }); + expect(collected[4].querySplitResponse).to.be.deep.equal({ + LastEvaluatedKey: { + pk: '6', + sk: 'thing', + }, + Item: { + pk: '6', + sk: 'thing', + name: 'thing 6', + }, + }); + expect(collected[4].emit).to.deep.equal({ + type: 'xyz', + raw: { + pk: '6', + sk: 'thing', + name: 'thing 6', + }, + tags: { + account: 'undefined', + region: 'us-west-2', + stage: 'undefined', + source: 'undefined', + functionname: 'undefined', + pipeline: 'job1-continued', + skip: true, + }, + }); + expect(collected[4].cursorUpdateRequest).to.be.undefined; + expect(collected[5].cursorUpdateRequest).to.deep.equal({ + Key: { + pk: 'job-2', + sk: 'job', + }, + ExpressionAttributeNames: { + '#cursor': 'cursor', + '#timestamp': 'timestamp', + }, + ExpressionAttributeValues: { + ':timestamp': 1572832694000, + ':cursor': { + pk: '6', + sk: 'thing', + }, + }, + UpdateExpression: 'SET #cursor = :cursor, #timestamp = :timestamp', + ReturnValues: 'ALL_NEW', + ConditionExpression: 'attribute_not_exists(#timestamp) OR #timestamp < :timestamp', + }); + // End second uow cursor processing + }) + .done(done); + }); + it('should stop', (done) => { const events = toDynamodbRecords([ {