@emartech/program-executor - npm Package Compare versions

Comparing version 3.12.0 to 3.12.1

package.json

@@ -23,3 +23,3 @@ {
     "@emartech/json-logger": "3.4.0",
-    "@emartech/pubsub-client-js": "^1.2.5",
+    "@emartech/pubsub-client-js": "^1.2.7",
     "@emartech/rabbitmq-client": "5.9.1",
@@ -43,3 +43,3 @@ "camelcase-keys": "^6.2.2",
   },
-  "version": "3.12.0"
+  "version": "3.12.1"
 }

src/index-pubsub-e2e.test.js

@@ -14,14 +14,52 @@ 'use strict';
 
+class SuccessfulJob {
+  static get name() {
+    return 'successful_job';
+  }
+
+  static create(programData) {
+    return new SuccessfulJob(programData);
+  }
+
+  // eslint-disable-next-line no-unused-vars
+  constructor(programData) {
+    /// ... initialize member variables based on programData if needed
+  }
+
+  async execute(message, jobDataHandler) {
+    // eslint-disable-next-line no-unused-vars
+    const jobSpecificData = await jobDataHandler.get();
+  }
+}
+
+class ProblematicJob {
+  static get name() {
+    return 'problematic_job';
+  }
+
+  static create(programData) {
+    return new ProblematicJob(programData);
+  }
+
+  // eslint-disable-next-line no-unused-vars
+  constructor(programData) {
+  }
+
+  // eslint-disable-next-line no-unused-vars
+  async execute(message, jobDataHandler) {
+    throw new Error('Some error happened in the job');
+  }
+}
+
+
 const testJobLibrary = {
   firstJob: {},
-  secondJob: {}
+  secondJob: {},
+  successful_job: SuccessfulJob,
+  problematic_job: ProblematicJob
 };
 
-const waitForMessages = (timeout) => {
-  let timeOutLength = timeout;
-  return new Promise((resolve) => setTimeout(resolve, timeOutLength));
-};
-
 describe('ProgramExecutor', function () {
   let config;
+  let configFlowControl;
   let dbConnection;
@@ -52,3 +90,2 @@ let mockConstructorSpy;
     jest.clearAllMocks();
-    jest.resetModules();
   });
@@ -65,2 +102,12 @@
     };
+
+    configFlowControl = {
+      knex: dbConnection,
+      tableName: 'programs',
+      topicName: topicName,
+      projectId: 'some-project-id',
+      pubSubMaxExtensionMinutes: 20,
+      pubSubMinAckDeadline: 10,
+      pubSubMaxAckDeadline: {"millis":600000}
+    };
     const result = await e2eTestReuse.beforeEach(topicName);
@@ -73,3 +120,3 @@ mockConstructorSpy = result.mockConstructorSpy;
   describe('#Program Executor', function () {
-    it('should execute program and maintain status in DB', async function () {
+    it('should execute program and maintain status in DB: two skipped jobs', async function () {
       const runId = await ProgramExecutor.create(config).createProgram(
@@ -85,3 +132,3 @@ {
 
-      await waitForMessages(1000);
+      await e2eTestReuse.waitForMessages(1000);
 
@@ -97,4 +144,72 @@ const result = await dbConnection('programs').where({ run_id: runId }).first();
     });
+
+        it('should execute program and maintain status in DB: two skipped jobs (with flow control config)', async function () {
+      const runId = await ProgramExecutor.create(configFlowControl).createProgram(
+        {
+          programData: {
+            customerId: customerId
+          },
+          jobs: ['current_program', 'next_program']
+        }
+      );
+      await ProgramExecutor.create(configFlowControl).processPrograms(testJobLibrary);
+
+      await e2eTestReuse.waitForMessages(1000);
+
+      const result = await dbConnection('programs').where({ run_id: runId }).first();
+
+      expect(result.run_id).toBe(runId);
+      expect(result.jobs).toEqual(['current_program', 'next_program']);
+      expect(result.step).toEqual(1);
+      expect(result.errored_at).toEqual(null);
+      expect(result.finished_at).not.toEqual(null);
+      expect(result.program_data).toEqual({ customerId: 1234 });
+    });
+
+    it('should execute program and maintain status in DB: one executed job', async function () {
+      const runId = await ProgramExecutor.create(config).createProgram(
+        {
+          programData: {
+            customerId: customerId
+          },
+          jobs: ['successful_job']
+        }
+      );
+      await ProgramExecutor.create(config).processPrograms(testJobLibrary);
+
+      await e2eTestReuse.waitForMessages(1000);
+
+      const result = await dbConnection('programs').where({ run_id: runId }).first();
+
+      expect(result.run_id).toBe(runId);
+      expect(result.jobs).toEqual(['successful_job']);
+      expect(result.step).toEqual(0);
+      expect(result.errored_at).toEqual(null);
+      expect(result.finished_at).not.toEqual(null);
+      expect(result.program_data).toEqual({ customerId: 1234 });
+    });
+
+    it('should execute program and maintain status in DB: one failed job', async function () {
+      const runId = await ProgramExecutor.create(config).createProgram(
+        {
+          programData: {
+            customerId: customerId
+          },
+          jobs: ['problematic_job']
+        }
+      );
+      await ProgramExecutor.create(config).processPrograms(testJobLibrary);
+
+      await e2eTestReuse.waitForMessages(1000);
+
+      const result = await dbConnection('programs').where({ run_id: runId }).first();
+
+      expect(result.run_id).toBe(runId);
+      expect(result.jobs).toEqual(['problematic_job']);
+      expect(result.step).toEqual(0);
+      expect(result.errored_at).not.toEqual(null);
+      expect(result.finished_at).toEqual(null);
+      expect(result.program_data).toEqual({ customerId: 1234 });
+    });
   });
-
 });

src/index-pubsub.js

@@ -17,2 +17,5 @@ 'use strict';
    * @param {string} config.topicName - Topic name to publish to
+   * @param {string} config.pubSubMaxExtensionMinutes - Pubsub lease management: max extension minutes
+   * @param {string} config.pubSubMinAckDeadline - Pubsub lease management: Min Ack Dead line
+   * @param {string} config.pubSubMaxAckDeadline - Pubsub lease management: Max Ack Dead line
    */
@@ -52,2 +55,5 @@ constructor(config) {
           MaxMessages: 1,
+          MaxExtensionMinutes:this._config.pubSubMaxExtensionMinutes || 60,
+          MinAckDeadline:this._config.pubSubMinAckDeadline || undefined,
+          MaxAckDeadline:this._config.pubSubMaxAckDeadline || undefined,      
           onMessage: async (message) => {
@@ -54,0 +60,0 @@ try {

src/index-pubsub.test.js

@@ -23,3 +23,6 @@ 'use strict';
       topicName: 'program-executor',
-      projectId: 'amqp://guest:guest@localhost:9999'
+      projectId: 'amqp://guest:guest@localhost:9999',
+      pubSubMaxExtensionMinutes: 60,
+      pubSubMinAckDeadline: {'millis':600000},
+      pubSubMaxAckDeadline: {'millis':600000}
     };
@@ -86,2 +89,5 @@
         MaxStreams: 1,
+        MaxExtensionMinutes: 60,
+        MinAckDeadline:{'millis':600000},
+        MaxAckDeadline:{'millis':600000},
         logger: 'program-executor-consumer' }));
@@ -88,0 +94,0 @@ });

New alerts

Environment variable access

Supply chain risk

Package accesses environment variables, which may be a sign of credential stuffing or data theft.

Found 7 instances in 1 package

Fixed alerts

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Environment variable access

Supply chain risk

Package accesses environment variables, which may be a sign of credential stuffing or data theft.

Found 7 instances in 1 package

Improved metrics

Total package byte prevSize

106579

4.31%

Lines of code

2364

4.74%

Number of medium supply chain risk alerts

0

-100%

Dependency changes

• Updated

  @emartech/pubsub-client-js@^1.2.7