enabled pipelines env var and npm-oidc

.github/workflows/cd.yml

@@ -5,6 +5,10 @@ on:
   # push:
   #   branches: [ master ]
 
+permissions:
+  id-token: write  # Required for OIDC
+  contents: read
+
 jobs:
   publish:
     runs-on: ubuntu-latest
@@ -14,11 +18,12 @@ jobs:
         with:
           node-version: 18.x
           registry-url: https://registry.npmjs.org/
+      # Ensure npm 11.5.1 or later is installed
+      - name: Update npm
+        run: npm install -g npm@latest          
       - run: npm ci
       - run: npm test
       - run: npm publish
-        env:
-          NODE_AUTH_TOKEN: ${{secrets.NODE_AUTH_TOKEN}}
       - uses: sergeysova/jq-action@v2
         name: Derive Version
         id: version

package.json

@@ -1,6 +1,6 @@
 {
   "name": "aws-lambda-stream",
-  "version": "1.1.13",
+  "version": "1.1.14",
   "description": "Create stream processors with AWS Lambda functions.",
   "keywords": [
     "aws",

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(',').map((pipeline) => pipeline.trim());
+  const disabledPipelines = (opt.DISABLED_PIPELINES || process.env.DISABLED_PIPELINES)?.split(',').map((pipeline) => pipeline.trim());
   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 '; // extra space on purpose
+    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;