lint and mutex.wrap

Author: ryanhs

.eslintrc.js

@@ -0,0 +1,65 @@
+module.exports = {
+  env: {
+    commonjs: true,
+    es6: true,
+    node: true,
+  },
+  extends: ['airbnb-base', 'plugin:jest/recommended', 'plugin:prettier/recommended'],
+  globals: {
+    Atomics: 'readonly',
+    SharedArrayBuffer: 'readonly',
+    fetch: 'readonly',
+    fetchMock: 'readonly'
+  },
+  parserOptions: {
+    ecmaVersion: 2018,
+  },
+  plugins: ['jest'],
+  rules: {
+    'camelcase': 'off',
+    'no-console': 'warn',
+    'no-trailing-spaces': 'off',
+    'global-require': 'off',
+    'no-underscore-dangle': 'off', // mongoId
+    'padded-blocks': 'off',
+    'no-unused-vars': 'warn',
+    'no-param-reassign': ['error', { props: false }],
+    'radix': 'warn',
+    'max-len': [
+      'error',
+      { code: 120, comments: 200, ignoreTrailingComments: true, ignoreUrls: true },
+    ],
+    'prettier/prettier': ['error'],
+    "jest/no-focused-tests": "warn",
+    "jest/no-standalone-expect": "warn",
+  },
+  overrides: [
+    {
+      files: ['tests/**/*.js'],
+      rules: {
+        'no-unused-vars': 'off',
+        'no-console': 'off',
+        'prefer-const': 'off',
+      },
+    },
+    {
+      files: ['*.md'],
+      rules: {
+        'arrow-body-style': 0,
+        'consistent-return': 0,
+        'flowtype/require-valid-file-annotation': 0,
+        'import/no-extraneous-dependencies': 0,
+        'import/no-unresolved': 0,
+        'jest/no-focused-tests': 0,
+        'jest/no-identical-title': 0,
+        'jest/valid-expect': 0,
+        'no-undef': 0,
+        'no-unused-vars': 0,
+        'prettier/prettier': 0,
+        'react/jsx-no-undef': 0,
+        'react/react-in-jsx-scope': 0,
+        'sort-keys': 0,
+      },
+    },
+  ],
+};

.eslintrc.json

@@ -1,5 +1,5 @@
 {
-  "printWidth": 100,
+  "printWidth": 120,
   "singleQuote": true,
   "trailingComma": "all"
 }

.prettierrc

@@ -58,10 +58,7 @@ const makeRequest = ({ sdk, method, client, remoteServiceName, logger }) => (con
         if (throwError) {
           // if there is more information
           if (Array.isArray(err.graphQLErrors) && err.graphQLErrors.length > 0) {
-            throw flaverr(
-              _.pick(err.graphQLErrors[0], ['code', 'httpStatusCode', 'locations', 'path']),
-              err,
-            );
+            throw flaverr(_.pick(err.graphQLErrors[0], ['code', 'httpStatusCode', 'locations', 'path']), err);
           }
 
           // else just throw error

src/graphql/connect.js

@@ -157,12 +157,7 @@ async function disableGraphql(name = 'default') {
 
 // ---------------------------------------------------------------------------
 
-async function bootstrap(
-  environment = 'production',
-  configOverrides = {},
-  useEnv = true,
-  overwriteGlobal = false,
-) {
+async function bootstrap(environment = 'production', configOverrides = {}, useEnv = true, overwriteGlobal = false) {
   // create new instance
   const instance = _.cloneDeep(defaultInstance);
 

src/index.js

@@ -1,4 +1,5 @@
 const flaverr = require('flaverr');
+const bluebird = require('bluebird');
 const DriverMemory = require('./drivers/memory');
 const DriverRedis = require('./drivers/redis');
 
@@ -61,6 +62,18 @@ module.exports = async ({ configs, sdk }) => {
     acquire: (args) => connection.acquire(args),
     destroy: () => connection.destroyInstance(),
 
+    wrap: async (args, fn) => {
+      const unlock = await connection.acquire(args);
+
+      return bluebird
+        .resolve(fn())
+        .tap(unlock)
+        .catch(async (err) => {
+          await unlock();
+          throw err;
+        });
+    },
+
     // dig deeper
     connection,
   };

src/mutex/connect.js

@@ -24,18 +24,7 @@ module.exports = {
   friendlyName: 'createInstance',
   description: 'Create stream driver with kafka',
 
-  fn(
-    {
-      sdk,
-      connectionString,
-      sslEnable = false,
-      sslRejectUnauthorized,
-      sslCaFile,
-      sslKeyFile,
-      sslCertFile,
-    },
-    exits,
-  ) {
+  fn({ sdk, connectionString, sslEnable = false, sslRejectUnauthorized, sslCaFile, sslKeyFile, sslCertFile }, exits) {
     // notify to boot
     const log = sdk.log.child({
       service: 'stream',

src/streams/drivers/kafka/machines/createInstance.js

@@ -95,24 +95,22 @@ describe('check', () => {
 
     const s = makeObservable({ client, sdk, channel: channel2 })
       .now()
-      .observable
-      .subscribe({ next: () => mock() });
+      .observable.subscribe({ next: () => mock() });
 
-    await Promise.delay(25)
+    await Promise.delay(25);
     await sendSomething(channel1);
-    await Promise.delay(25)
+    await Promise.delay(25);
     await sendSomething(channel2);
 
     // make sure in next 10s
-    await (new Promise(resolve => {
+    await new Promise((resolve) => {
       setTimeout(() => {
-        expect(mock).not.toHaveBeenCalled()
+        expect(mock).not.toHaveBeenCalled();
 
         s.unsubscribe();
         resolve();
-      }, 10000)
-    }))
-
+      }, 10000);
+    });
   });
 
   it('fromBeginning ok', async () => {

src/streams/drivers/kafka/machines/makeObservable.test.js

@@ -0,0 +1,95 @@
+const Promise = require('bluebird');
+const uuid = require('uuid').v4;
+
+const bootstrap = require('./_bootstrapSdk');
+
+jest.useRealTimers();
+jest.setTimeout(500000);
+let sdk;
+
+beforeAll(async () => {
+  sdk = await bootstrap({
+    APP_NAME: 'jest',
+    LOG_DEBUG_ENABLE: true,
+    MUTEX_DRIVER: 'memory',
+  });
+  await sdk.enable_mutex();
+});
+
+afterAll(() => jest.clearAllTimers());
+
+describe('check wrap returns', () => {
+  it('wrap mutex ok on resolves', async () => {
+    await expect(sdk.mutex.wrap({ key: 'trx' }, () => 1)).resolves.not.toThrow();
+  });
+
+  it('wrap mutex ok on rejects', async () => {
+    await expect(
+      sdk.mutex.wrap({ key: 'trx' }, () => {
+        throw new Error('aha');
+      }),
+    ).rejects.toThrow(/aha/);
+  });
+});
+
+describe('check wrap no race', () => {
+  it('no races', async () => {
+    const mutexKey = uuid();
+
+    const beforeAllAt = Date.now();
+    const loggerTable = [];
+
+    const run = async (workerId, workDuration, shouldRunAfter) => {
+      const createdAt = Date.now();
+      let startedAt;
+      let finishedAt;
+
+      await sdk.mutex.wrap({ key: mutexKey }, async () => {
+        startedAt = Date.now();
+        await Promise.delay(workDuration);
+        loggerTable.push({
+          workerId,
+          waitInSeconds: workDuration / 1000,
+          startedAt: new Date(startedAt).toISOString().split('T')[1],
+          shouldRunAfter: new Date(beforeAllAt + shouldRunAfter).toISOString().split('T')[1],
+        });
+        finishedAt = Date.now();
+      });
+
+      // its ok?
+      expect(createdAt).toBeGreaterThan(beforeAllAt);
+      expect(startedAt).toBeGreaterThan(beforeAllAt);
+      expect(finishedAt).toBeGreaterThan(beforeAllAt);
+
+      // timer in order
+      expect(startedAt).toBeGreaterThan(createdAt);
+      expect(finishedAt).toBeGreaterThan(startedAt);
+
+      // outside timer
+      expect(startedAt).toBeGreaterThanOrEqual(beforeAllAt + shouldRunAfter);
+
+      return {
+        createdAt,
+        startedAt,
+        finishedAt,
+      };
+    };
+
+    // mutex not run in order, then we only make sure if its run after certain time
+    await Promise.all([
+      // #1 batch
+      // called At                      id   run in (ms)    startedAt should be >=
+      Promise.delay(1).then(() => run('1.1', 2000, 1)),
+      Promise.delay(1000).then(() => run('1.2', 2000, 2000)), // batch #1.2 will be randomly assigned
+      Promise.delay(2000).then(() => run('1.3', 2000, 2000)), // batch #1.2 will be randomly assigned
+
+      // #3 batch
+      // called At                      id   run in (ms)    startedAt should be >=
+      Promise.delay(8000).then(() => run('2.1', 2000, 8000)),
+      Promise.delay(9000).then(() => run('2.2', 2000, 9000)),
+      Promise.delay(10000).then(() => run('2.3', 2000, 9000)),
+    ]);
+
+    // console.table(loggerTable)
+  });
+});

tests/mutex/wrap.test.js

@@ -21,21 +21,15 @@ describe('check', () => {
     expect(sdk.zipkinTracer.local('static int 1', () => 1)).toBe(1);
     expect(sdk.zipkinTracer.local('static int 2', () => 2)).toBe(2);
 
-    const spansLength = sdk.log.ringBuffer.records.filter(({ log }) => log === 'zipkin-span')
-      .length;
+    const spansLength = sdk.log.ringBuffer.records.filter(({ log }) => log === 'zipkin-span').length;
     expect(spansLength).toBe(2);
   });
 
   it('local process: promise ', async () => {
-    await expect(
-      sdk.zipkinTracer.local('promise int 1', () => Promise.delay(200, 1)),
-    ).resolves.toBe(1);
-    await expect(
-      sdk.zipkinTracer.local('promise int 2', () => Promise.delay(200, 2)),
-    ).resolves.toBe(2);
+    await expect(sdk.zipkinTracer.local('promise int 1', () => Promise.delay(200, 1))).resolves.toBe(1);
+    await expect(sdk.zipkinTracer.local('promise int 2', () => Promise.delay(200, 2))).resolves.toBe(2);
 
-    const spansLength = sdk.log.ringBuffer.records.filter(({ log }) => log === 'zipkin-span')
-      .length;
+    const spansLength = sdk.log.ringBuffer.records.filter(({ log }) => log === 'zipkin-span').length;
     expect(spansLength).toBe(2);
   });
 });

tests/tracers/zipkin/common-memory.test.js

@@ -58,9 +58,7 @@ describe('check cache oke', () => {
       },
     });
 
-    expect(cache.redisCache.store.getClient().options.url).toBe(
-      process.env.REDIS_CONNECTION_STRING,
-    );
+    expect(cache.redisCache.store.getClient().options.url).toBe(process.env.REDIS_CONNECTION_STRING);
 
     return cache.redisCache.store.getClient().quit();
   });