Update materialize flavor.

Author: petermyers

src/flavors/materialize.js

@@ -26,6 +26,9 @@ export const materialize = (rule) => (s) => s // eslint-disable-line import/pref
   .map(toUpdateRequest(rule))
   .parallel(rule.parallel || Number(process.env.PARALLEL) || 4)
 
+  .map(toFallbackUpdateRequest(rule))
+  .parallel(rule.parallel || Number(process.env.PARALLEL) || 4)
+
   .through(updateDynamoDB(rule))
 
   .tap(printEndPipeline);
@@ -37,3 +40,8 @@ const toUpdateRequest = (rule) => faultyAsyncStream(async (uow) => ({
   ...uow,
   updateRequest: await faultify(rule.toUpdateRequest)(uow, rule),
 }));
+
+const toFallbackUpdateRequest = (rule) => faultyAsyncStream(async (uow) => ({
+  ...uow,
+  fallbackUpdateRequest: rule.toFallbackUpdateRequest ? await faultify(rule.toFallbackUpdateRequest)(uow, rule) : undefined,
+}));

test/unit/flavors/materialize.test.js

@@ -2,6 +2,7 @@ import 'mocha';
 import { expect } from 'chai';
 import sinon from 'sinon';
 
+import { cloneDeep } from 'lodash';
 import {
   initialize, initializeFrom,
   ttl,
@@ -92,6 +93,99 @@ describe('flavors/materialize.js', () => {
       })
       .done(done);
   });
+
+  it('should optionally call the fallback update request', (done) => {
+    const events = toKinesisRecords([
+      {
+        type: 'm1',
+        timestamp: 1548967022000,
+        thing: {
+          id: '1',
+          name: 'Thing One',
+          description: 'This is thing one',
+        },
+      },
+      {
+        type: 'split',
+        timestamp: 1548967022000,
+        root: {
+          things: [{
+            id: '2',
+            name: 'Thing One',
+            description: 'This is thing one',
+          }, {
+            id: '3',
+            name: 'Thing One',
+            description: 'This is thing one',
+          }],
+        },
+      },
+    ]);
+
+    const ruleWithFallbackUpdateRequest = cloneDeep(rules[0]);
+    ruleWithFallbackUpdateRequest.toFallbackUpdateRequest = (uow) => ({
+      Key: {
+        pk: uow.event.thing.id,
+        sk: 'thing',
+      },
+      ...updateExpression({
+        fallbackUpdate: true,
+      }),
+    });
+
+    initialize({
+      ...initializeFrom([ruleWithFallbackUpdateRequest]),
+    })
+      .assemble(fromKinesis(events), false)
+      .collect()
+      // .tap((collected) => console.log(JSON.stringify(collected, null, 2)))
+      .tap((collected) => {
+        expect(collected.length).to.equal(1);
+        expect(collected[0].pipeline).to.equal('mv1');
+        expect(collected[0].event.type).to.equal('m1');
+        expect(collected[0].updateRequest).to.deep.equal({
+          Key: {
+            pk: '1',
+            sk: 'thing',
+          },
+          ExpressionAttributeNames: {
+            '#id': 'id',
+            '#name': 'name',
+            '#description': 'description',
+
+            '#discriminator': 'discriminator',
+            '#ttl': 'ttl',
+            '#timestamp': 'timestamp',
+          },
+          ExpressionAttributeValues: {
+            ':id': '1',
+            ':name': 'Thing One',
+            ':description': 'This is thing one',
+            ':discriminator': 'thing',
+            ':ttl': 1549053422,
+            ':timestamp': 1548967022000,
+          },
+          UpdateExpression: 'SET #id = :id, #name = :name, #description = :description, #discriminator = :discriminator, #ttl = :ttl, #timestamp = :timestamp',
+          ReturnValues: 'ALL_NEW',
+          ConditionExpression: 'attribute_not_exists(#timestamp) OR #timestamp < :timestamp',
+        });
+        expect(collected[0].fallbackUpdateRequest).to.deep.equal({
+          Key: {
+            pk: '1',
+            sk: 'thing',
+          },
+          ExpressionAttributeNames: {
+            '#fallbackUpdate': 'fallbackUpdate',
+          },
+          ExpressionAttributeValues: {
+            ':fallbackUpdate': true,
+          },
+          UpdateExpression: 'SET #fallbackUpdate = :fallbackUpdate',
+          ReturnValues: 'ALL_NEW',
+        });
+      })
+      .done(done);
+  });
 });
 
 const toUpdateRequest = (uow) => ({

test/unit/metrics/index.test.js

@@ -187,17 +187,17 @@ describe('metrics/index.js', () => {
             count: 3,
           },
           'p1|save|stream.pipeline.io.wait.time': {
-            average: 20,
-            min: 14,
-            max: 26,
-            sum: 60,
+            average: 22,
+            min: 16,
+            max: 28,
+            sum: 66,
             count: 3,
           },
           'p1|save|stream.pipeline.io.time': {
-            average: 8,
-            min: 8,
-            max: 8,
-            sum: 24,
+            average: 6,
+            min: 6,
+            max: 6,
+            sum: 18,
             count: 3,
           },
           'p1|stream.pipeline.time': {
@@ -229,17 +229,17 @@ describe('metrics/index.js', () => {
             count: 1,
           },
           'p2|get|stream.pipeline.io.wait.time': {
-            average: 8,
-            min: 8,
-            max: 8,
-            sum: 8,
+            average: 2,
+            min: 2,
+            max: 2,
+            sum: 2,
             count: 1,
           },
           'p2|get|stream.pipeline.io.time': {
-            average: 14,
-            min: 14,
-            max: 14,
-            sum: 14,
+            average: 20,
+            min: 20,
+            max: 20,
+            sum: 20,
             count: 1,
           },
           'p2|publish|stream.pipeline.io.wait.time': {