allow overriding the cdc event field for cdc short circuiting

Author: codyjmoody

src/flavors/cdc.js

@@ -29,7 +29,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);
@@ -55,12 +59,11 @@ const toGetRequest = (rule) => faulty((uow) => ({
       : undefined,
 }));
 
-const toEvent = (rule) => faultyAsyncStream(async (uow) => (!rule.toEvent
-  ? uow
-  : ({
+const toEvent = (rule) => faultyAsyncStream(async (uow) => (!rule.toEvent // eslint-disable-line no-nested-ternary
+  ? uow : ({
     ...uow,
-    event: {
-      ...uow.event,
+    [rule.eventField || 'event']: ({
+      ...(!rule.eventField && 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,26 +86,40 @@ describe('flavors/cdc.js', () => {
       .collect()
       // .tap((collected) => console.log(JSON.stringify(collected, null, 2)))
       .tap((collected) => {
-        expect(collected.length).to.equal(2);
-        expect(collected[1].pipeline).to.equal('cdc1');
-        expect(collected[1].event.type).to.equal('thing-created');
-        expect(collected[1].event.thing).to.deep.equal({
+        expect(collected.length).to.equal(3);
+        expect(collected[0].pipeline).to.equal('cdc1');
+        expect(collected[0].event.type).to.equal('thing-created');
+        expect(collected[0].event.thing).to.deep.equal({
           id: '1',
           name: 'IlRoaW5nIE9uZSI=', // 'Thing One',
           description: 'This is thing one',
         });
-        expect(collected[1].event.tags).to.deep.equal({
+        expect(collected[0].event.tags).to.deep.equal({
           region: 'us-west-2',
           field1: 'v1',
           ...envTags('cdc1'),
           ...skipTag(),
         });
-        expect(collected[1].queryRequest).to.be.undefined;
+        expect(collected[0].queryRequest).to.be.undefined;
+
+        expect(collected[1].pipeline).to.equal('cdc2');
+        expect(collected[1].queryRequest).to.not.be.undefined;
+        expect(collected[1].queryResponse).to.not.be.undefined;
 
-        // this pipeline speeds ahead since it does 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;
+        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(),
+        });
       })
       .done(done);
   });
@@ -129,4 +159,14 @@ const rules = [
     flavor: cdc,
     eventType: 'x9',
   },
+  {
+    id: 'cdc3',
+    flavor: cdc,
+    toEvent,
+    eventField: 'emit',
+    eventType: /override-*/,
+    eem: {
+      fields: ['name'],
+    },
+  },
 ];