feat(sandboxed-process): support for esm files

.github/workflows/node.js.yml

@@ -44,7 +44,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [10.x, 12.x, 14.x, 16.x]
+        node-version: [12.x, 14.x, 16.x]
 
     steps:
       - name: Checkout repository and submodules

src/classes/child-processor.ts

@@ -24,7 +24,7 @@ export class ChildProcessor {
   public async init(processorFile: string): Promise<void> {
     let processor;
     try {
-      processor = require(processorFile);
+      processor = await import(processorFile);
     } catch (err) {
       this.status = ChildStatus.Errored;
       return childSend(process, {

src/classes/worker.ts

@@ -175,7 +175,7 @@ export class Worker<
         this.processFn = processor;
       } else {
         // SANDBOXED
-        const supportedFileTypes = ['.js', '.ts', '.flow', '.cjs'];
+        const supportedFileTypes = ['.js', '.ts', '.flow', '.cjs', '.mjs'];
         const processorFile =
           processor +
           (supportedFileTypes.includes(path.extname(processor)) ? '' : '.js');

tests/fixtures/fixture_processor.mjs

@@ -0,0 +1,11 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+import * as delay from './delay';
+
+export default function(/*job*/) {
+  return delay(500).then(() => {
+    return 42;
+  });
+};

tests/test_sandboxed_process.ts

@@ -11,6 +11,7 @@ import {
 import { beforeEach } from 'mocha';
 import { v4 } from 'uuid';
 import { delay, removeAllQueueData } from '../src/utils';
+import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';
 const { stdout, stderr } = require('test-console');
 
 describe('sandboxed process', () => {
@@ -71,7 +72,40 @@ describe('sandboxed process', () => {
     it('processes and completes', async () => {
       const processFile = __dirname + '/fixtures/fixture_processor.cjs';
       const worker = new Worker(queueName, processFile, {
-        autorun: false,
+        connection,
+        drainDelay: 1,
+      });
+
+      const completing = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async (job: Job, value: any) => {
+          try {
+            expect(job.data).to.be.eql({ foo: 'bar' });
+            expect(value).to.be.eql(42);
+            expect(Object.keys(worker['childPool'].retained)).to.have.lengthOf(
+              0,
+            );
+            expect(worker['childPool'].free[processFile]).to.have.lengthOf(1);
+            await worker.close();
+            resolve();
+          } catch (err) {
+            await worker.close();
+            reject(err);
+          }
+        });
+      });
+
+      worker.run();
+
+      await queue.add('foobar', { foo: 'bar' });
+
+      await completing;
+    });
+  });
+
+  describe('when processor file is .mjs (ESM)', () => {
+    it('processes and completes', async () => {
+      const processFile = __dirname + '/fixtures/fixture_processor.mjs';
+      const worker = new Worker(queueName, processFile, {
         connection,
         drainDelay: 1,
       });