Initial commit

node_modules/knex/lib/dialects/oracledb/utils.js

@@ -0,0 +1,208 @@
+const Utils = require('../oracle/utils');
+const { promisify } = require('util');
+const stream = require('stream');
+
+function BlobHelper(columnName, value) {
+  this.columnName = columnName;
+  this.value = value;
+  this.returning = false;
+}
+
+BlobHelper.prototype.toString = function () {
+  return '[object BlobHelper:' + this.columnName + ']';
+};
+
+/**
+ * @param stream
+ * @param {'string' | 'buffer'} type
+ */
+function readStream(stream, type) {
+  return new Promise((resolve, reject) => {
+    let data = type === 'string' ? '' : Buffer.alloc(0);
+
+    stream.on('error', function (err) {
+      reject(err);
+    });
+    stream.on('data', function (chunk) {
+      if (type === 'string') {
+        data += chunk;
+      } else {
+        data = Buffer.concat([data, chunk]);
+      }
+    });
+    stream.on('end', function () {
+      resolve(data);
+    });
+  });
+}
+
+const lobProcessing = function (stream) {
+  const oracledb = require('oracledb');
+
+  /**
+   * @type 'string' | 'buffer'
+   */
+  let type;
+
+  if (stream.type) {
+    // v1.2-v4
+    if (stream.type === oracledb.BLOB) {
+      type = 'buffer';
+    } else if (stream.type === oracledb.CLOB) {
+      type = 'string';
+    }
+  } else if (stream.iLob) {
+    // v1
+    if (stream.iLob.type === oracledb.CLOB) {
+      type = 'string';
+    } else if (stream.iLob.type === oracledb.BLOB) {
+      type = 'buffer';
+    }
+  } else {
+    throw new Error('Unrecognized oracledb lob stream type');
+  }
+  if (type === 'string') {
+    stream.setEncoding('utf-8');
+  }
+  return readStream(stream, type);
+};
+
+function monkeyPatchConnection(connection, client) {
+  // Connection is already monkey-patched
+  if (connection.executeAsync) {
+    return;
+  }
+
+  connection.commitAsync = function () {
+    return new Promise((commitResolve, commitReject) => {
+      this.commit(function (err) {
+        if (err) {
+          return commitReject(err);
+        }
+        commitResolve();
+      });
+    });
+  };
+  connection.rollbackAsync = function () {
+    return new Promise((rollbackResolve, rollbackReject) => {
+      this.rollback(function (err) {
+        if (err) {
+          return rollbackReject(err);
+        }
+        rollbackResolve();
+      });
+    });
+  };
+  const fetchAsync = promisify(function (sql, bindParams, options, cb) {
+    options = options || {};
+    options.outFormat = client.driver.OUT_FORMAT_OBJECT || client.driver.OBJECT;
+    if (!options.outFormat) {
+      throw new Error('not found oracledb.outFormat constants');
+    }
+    if (options.resultSet) {
+      connection.execute(
+        sql,
+        bindParams || [],
+        options,
+        function (err, result) {
+          if (err) {
+            if (Utils.isConnectionError(err)) {
+              connection.close().catch(function (err) {});
+              connection.__knex__disposed = err;
+            }
+            return cb(err);
+          }
+          const fetchResult = { rows: [], resultSet: result.resultSet };
+          const numRows = 100;
+          const fetchRowsFromRS = function (connection, resultSet, numRows) {
+            resultSet.getRows(numRows, function (err, rows) {
+              if (err) {
+                if (Utils.isConnectionError(err)) {
+                  connection.close().catch(function (err) {});
+                  connection.__knex__disposed = err;
+                }
+                resultSet.close(function () {
+                  return cb(err);
+                });
+              } else if (rows.length === 0) {
+                return cb(null, fetchResult);
+              } else if (rows.length > 0) {
+                if (rows.length === numRows) {
+                  fetchResult.rows = fetchResult.rows.concat(rows);
+                  fetchRowsFromRS(connection, resultSet, numRows);
+                } else {
+                  fetchResult.rows = fetchResult.rows.concat(rows);
+                  return cb(null, fetchResult);
+                }
+              }
+            });
+          };
+          fetchRowsFromRS(connection, result.resultSet, numRows);
+        }
+      );
+    } else {
+      connection.execute(
+        sql,
+        bindParams || [],
+        options,
+        function (err, result) {
+          if (err) {
+            // dispose the connection on connection error
+            if (Utils.isConnectionError(err)) {
+              connection.close().catch(function (err) {});
+              connection.__knex__disposed = err;
+            }
+            return cb(err);
+          }
+
+          return cb(null, result);
+        }
+      );
+    }
+  });
+  connection.executeAsync = function (sql, bindParams, options) {
+    // Read all lob
+    return fetchAsync(sql, bindParams, options).then(async (results) => {
+      const closeResultSet = () => {
+        return results.resultSet
+          ? promisify(results.resultSet.close).call(results.resultSet)
+          : Promise.resolve();
+      };
+
+      // Collect LOBs to read
+      const lobs = [];
+      if (results.rows) {
+        if (Array.isArray(results.rows)) {
+          for (let i = 0; i < results.rows.length; i++) {
+            // Iterate through the rows
+            const row = results.rows[i];
+            for (const column in row) {
+              if (row[column] instanceof stream.Readable) {
+                lobs.push({ index: i, key: column, stream: row[column] });
+              }
+            }
+          }
+        }
+      }
+
+      try {
+        for (const lob of lobs) {
+          // todo should be fetchAsString/fetchAsBuffer polyfill only
+          results.rows[lob.index][lob.key] = await lobProcessing(lob.stream);
+        }
+      } catch (e) {
+        await closeResultSet().catch(() => {});
+
+        throw e;
+      }
+
+      await closeResultSet();
+
+      return results;
+    });
+  };
+}
+
+Utils.BlobHelper = BlobHelper;
+Utils.monkeyPatchConnection = monkeyPatchConnection;
+module.exports = Utils;