enabled pipelines env var

Author: John Gilbert

src/pipelines/index.js

@@ -49,9 +49,13 @@ export const initializeFrom = (rules) => rules.reduce(
 );
 
 const assemble = (opt) => (head, includeFaultHandler = true) => {
-  const disabledPipelines = process.env.DISABLED_PIPELINES?.split(',');
+  const enabledPipelines = (opt.ENABLED_PIPELINES || process.env.ENABLED_PIPELINES)?.split(',');
+  const disabledPipelines = (opt.DISABLED_PIPELINES || process.env.DISABLED_PIPELINES)?.split(',');
   const keys = Object.keys(thePipelines)
-    .filter((k) => !disabledPipelines?.includes(k));
+    .filter((k) => !disabledPipelines?.includes(k))
+    .filter((k) =>
+      !enabledPipelines || enabledPipelines.length === 0
+      || enabledPipelines.includes(k));
 
   debug('assemble: %j', keys);
 

test/unit/pipelines/pipelines.test.js

@@ -12,9 +12,14 @@ import Connector from '../../../src/connectors/eventbridge';
 describe('pipelines/index.js', () => {
   beforeEach(() => {
     sinon.stub(Connector.prototype, 'putEvents').resolves({ FailedEntryCount: 0 });
+    delete process.env.ENABLED_PIPELINES;
     delete process.env.DISABLED_PIPELINES;
   });
-  afterEach(sinon.restore);
+  afterEach(() => {
+    delete process.env.ENABLED_PIPELINES;
+    delete process.env.DISABLED_PIPELINES;
+    sinon.restore();
+  });
 
   it('should invoke all pipelines', (done) => {
     let counter = 0;
@@ -46,6 +51,36 @@ describe('pipelines/index.js', () => {
       .done(done);
   });
 
+  it('should only run enabled pipelines - string', (done) => {
+    process.env.ENABLED_PIPELINES = 'p1,p1b';
+    let counter = 0;
+
+    const count = (uow) => {
+      uow.counter = counter++; // eslint-disable-line no-plusplus
+      return uow;
+    };
+
+    const events = toKinesisRecords([{
+      type: 't1',
+    }]);
+
+    initialize({
+      p1: (opt) => (s) => s.map(count),
+      p1a: (opt) => (s) => s.map(count),
+      p1b: (opt) => (s) => s.map(count),
+    })
+      .assemble(fromKinesis(events), false)
+      .collect()
+      .tap((collected) => {
+        // console.log(JSON.stringify(collected, null, 2));
+        expect(collected.length).to.equal(2);
+        expect(counter).to.equal(2);
+        expect(collected[0].pipeline).to.equal('p1b');
+        expect(collected[1].pipeline).to.equal('p1');
+      })
+      .done(done);
+  });
+
   it('should ignore disabled pipelines - string', (done) => {
     process.env.DISABLED_PIPELINES = 'p1a';
     let counter = 0;