allow overriding the cdc event field

Author: codyjmoody

src/flavors/cdc.js

@@ -24,7 +24,11 @@ export const cdc = (rule) => (s) => s // eslint-disable-line import/prefer-defau
   .map(toEvent(rule))
   .parallel(rule.parallel || Number(process.env.PARALLEL) || 4)
 
-  .through(encryptEvent(rule))
+  .through(encryptEvent({
+    sourceField: rule.eventField || 'event',
+    targetField: rule.eventField || 'event',
+    ...rule,
+  }))
   .through(rule.publish(rule))
 
   .tap(printEndPipeline);
@@ -50,12 +54,16 @@ const toGetRequest = (rule) => faulty((uow) => ({
       : undefined,
 }));
 
-const toEvent = (rule) => faultyAsyncStream(async (uow) => (!rule.toEvent
-  ? uow
-  : ({
-    ...uow,
-    event: {
-      ...uow.event,
-      ...await faultify(rule.toEvent)(uow, rule),
-    },
-  })));
+const toEvent = (rule) => faultyAsyncStream(async (uow) => (!rule.toEvent // eslint-disable-line no-nested-ternary
+  ? uow : typeof rule.eventField === 'string' && rule.eventField !== 'event'
+    ? ({
+      ...uow,
+      [rule.eventField]: await faultify(rule.toEvent)(uow, rule),
+    })
+    : ({
+      ...uow,
+      event: {
+        ...uow.event,
+        ...await faultify(rule.toEvent)(uow, rule),
+      },
+    })));

test/unit/flavors/cdc.test.js

@@ -61,6 +61,22 @@ describe('flavors/cdc.js', () => {
           timestamp: 1548967022000,
         },
       },
+      {
+        timestamp: 1572832690,
+        keys: {
+          pk: '1',
+          sk: 'override',
+        },
+        newImage: {
+          pk: '1',
+          sk: 'override',
+          discriminator: 'override',
+          name: 'Override One',
+          description: 'This is override one',
+          ttl: 1549053422,
+          timestamp: 1548967022000,
+        },
+      },
     ]);
 
     initialize({
@@ -70,7 +86,7 @@ describe('flavors/cdc.js', () => {
       .collect()
       // .tap((collected) => console.log(JSON.stringify(collected, null, 2)))
       .tap((collected) => {
-        expect(collected.length).to.equal(2);
+        expect(collected.length).to.equal(3);
         expect(collected[1].pipeline).to.equal('cdc1');
         expect(collected[1].event.type).to.equal('thing-created');
         expect(collected[1].event.thing).to.deep.equal({
@@ -86,7 +102,22 @@ describe('flavors/cdc.js', () => {
         });
         expect(collected[1].queryRequest).to.be.undefined;
 
-        // this pipeline speeds ahead since it does less async
+        expect(collected[2].pipeline).to.equal('cdc3');
+        expect(collected[2].event.type).to.equal('override-created');
+        expect(collected[2].event.thing).to.be.undefined;
+        expect(collected[2].emit.thing).to.deep.equal({
+          id: '1',
+          name: 'Ik92ZXJyaWRlIE9uZSI=', // 'Override One'
+          description: 'This is override one',
+        });
+        expect(collected[2].emit.tags).to.deep.equal({
+          region: 'us-west-2',
+          field1: 'v1',
+          ...envTags('cdc3'),
+          ...skipTag(),
+        });
+
+        // this pipeline speeds ahead since they do less async
         expect(collected[0].pipeline).to.equal('cdc2');
         expect(collected[0].queryRequest).to.not.be.undefined;
         expect(collected[0].queryResponse).to.not.be.undefined;
@@ -129,4 +160,14 @@ const rules = [
     flavor: cdc,
     eventType: 'x9',
   },
+  {
+    id: 'cdc3',
+    flavor: cdc,
+    toEvent,
+    eventField: 'emit',
+    eventType: /override-*/,
+    eem: {
+      fields: ['name'],
+    },
+  },
 ];