Merge pull request #11 from topcoder-platform/transaction

Identify and combine same actions in a single mutex

Author: callmekatootie

src/common/helper.js

@@ -13,6 +13,7 @@ AWS.config.region = config.ES.AWS_REGION
 
 // Elasticsearch client
 let esClient
+
 // Mutex to ensure that only one elasticsearch action is carried out at any given time
 const esClientMutex = new Mutex()
 
@@ -55,20 +56,20 @@ async function getESClient () {
     })
   }
 
-  // Patch the transport to enable mutex
-  esClient.transport.originalRequest = esClient.transport.request
-  esClient.transport.request = async (params) => {
-    const release = await esClientMutex.acquire()
-    try {
-      return await esClient.transport.originalRequest(params)
-    } finally {
-      release()
-    }
-  }
-
   return esClient
 }
 
+/**
+ * Wraps original get es client function
+ * to control access to elasticsearch using a mutex
+ */
+async function getESClientWrapper () {
+  const client = await getESClient()
+  const release = await esClientMutex.acquire()
+
+  return { client, release }
+}
+
 /**
  * Function to valid require keys
  * @param {Object} payload validated object
@@ -86,17 +87,25 @@ function validProperties (payload, keys) {
 /**
  * Function to get user from es
  * @param {String} userId
- * @param {Boolean} sourceOnly
+ * @param {Boolean} isTransaction Is this part of a transaction?
  * @returns {Object} user
  */
-async function getUser (userId, sourceOnly = true) {
-  const client = await getESClient()
+async function getUser (userId, isTransaction = false) {
+  const { client, release } = await getESClientWrapper()
 
-  if (sourceOnly) {
-    return client.getSource({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
-  }
+  try {
+    const user = await client.get({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
+
+    if (isTransaction) {
+      return { seqNo: user._seq_no, primaryTerm: user._primary_term, user: user._source, release }
+    }
 
-  return client.get({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
+    return { seqNo: user._seq_no, primaryTerm: user._primary_term, user: user._source }
+  } finally {
+    if (!isTransaction) {
+      release()
+    }
+  }
 }
 
 /**
@@ -105,43 +114,92 @@ async function getUser (userId, sourceOnly = true) {
  * @param {Number} seqNo
  * @param {Number} primaryTerm
  * @param {Object} body
+ * @param {Boolean} isTransaction If this is part of a transaction, it will not attempt to release
  */
-async function updateUser (userId, body, seqNo, primaryTerm) {
-  const client = await getESClient()
-  await client.update({
-    index: config.get('ES.USER_INDEX'),
-    type: config.get('ES.USER_TYPE'),
-    id: userId,
-    body: { doc: body },
-    if_seq_no: seqNo,
-    if_primary_term: primaryTerm
-  })
+async function updateUser (userId, body, seqNo, primaryTerm, isTransaction = false) {
+  let client, release
+
+  if (isTransaction) {
+    client = await getESClient()
+  } else {
+    const esClient = await getESClientWrapper()
+    client = esClient.client
+    release = esClient.release
+  }
+
+  try {
+    await client.update({
+      index: config.get('ES.USER_INDEX'),
+      type: config.get('ES.USER_TYPE'),
+      id: userId,
+      body: { doc: body },
+      if_seq_no: seqNo,
+      if_primary_term: primaryTerm
+    })
+  } finally {
+    if (!isTransaction) {
+      release()
+    }
+  }
 }
 
 /**
  * Function to get org from es
  * @param {String} organizationId
+ * @param {Boolean} isTransaction Is this part of a transaction?
  * @returns {Object} organization
  */
-async function getOrg (organizationId) {
-  const client = await getESClient()
-  return client.getSource({ index: config.get('ES.ORGANIZATION_INDEX'), type: config.get('ES.ORGANIZATION_TYPE'), id: organizationId })
+async function getOrg (organizationId, isTransaction = false) {
+  const { client, release } = await getESClientWrapper()
+
+  try {
+    const org = await client.get({ index: config.get('ES.ORGANIZATION_INDEX'), type: config.get('ES.ORGANIZATION_TYPE'), id: organizationId })
+
+    if (isTransaction) {
+      return { seqNo: org._seq_no, primaryTerm: org._primary_term, org: org._source, release }
+    }
+
+    return { seqNo: org._seq_no, primaryTerm: org._primary_term, org: org._source }
+  } finally {
+    if (!isTransaction) {
+      release()
+    }
+  }
 }
 
 /**
  * Function to update es organization
  * @param {String} organizationId
  * @param {Object} body
+ * @param {Number} seqNo
+ * @param {Number} primaryTerm
+ * @param {Boolean} isTransaction If this is part of a transaction, it will not attempt to lock
  */
-async function updateOrg (organizationId, body) {
-  const client = await getESClient()
-  await client.update({
-    index: config.get('ES.ORGANIZATION_INDEX'),
-    type: config.get('ES.ORGANIZATION_TYPE'),
-    id: organizationId,
-    body: { doc: body },
-    refresh: 'true'
-  })
+async function updateOrg (organizationId, body, seqNo, primaryTerm, isTransaction = false) {
+  let client, release
+
+  if (isTransaction) {
+    client = await getESClient()
+  } else {
+    const esClient = await getESClientWrapper()
+    client = esClient.client
+    release = esClient.release
+  }
+
+  try {
+    await client.update({
+      index: config.get('ES.ORGANIZATION_INDEX'),
+      type: config.get('ES.ORGANIZATION_TYPE'),
+      id: organizationId,
+      body: { doc: body },
+      if_seq_no: seqNo,
+      if_primary_term: primaryTerm
+    })
+  } finally {
+    if (!isTransaction) {
+      release()
+    }
+  }
 }
 
 /**
@@ -158,7 +216,7 @@ function getErrorWithStatus (message, statusCode) {
 
 module.exports = {
   getKafkaOptions,
-  getESClient,
+  getESClientWrapper,
   validProperties,
   getUser,
   updateUser,