Skip to content

Commit

Permalink
refactor(migrator): refactor batchInsert from bluebird
Browse files Browse the repository at this point in the history
  • Loading branch information
maxeljkin committed Oct 10, 2019
1 parent f782348 commit 6de2633
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 42 deletions.
75 changes: 33 additions & 42 deletions lib/util/batchInsert.js
@@ -1,5 +1,5 @@
const { isNumber, chunk, flatten } = require('lodash');
const Bluebird = require('bluebird');
const delay = require('./delay');

module.exports = function batchInsert(
client,
Expand All @@ -11,60 +11,51 @@ module.exports = function batchInsert(
let autoTransaction = true;
let transaction = null;

const getTransaction = () =>
new Bluebird((resolve, reject) => {
if (transaction) {
autoTransaction = false;
return resolve(transaction);
}
const getTransaction = async () => {
if (transaction) {
autoTransaction = false;
return transaction;
}

autoTransaction = true;
client.transaction(resolve).catch(reject);
});
autoTransaction = true;
return client.transaction();
};

const wrapper = Object.assign(
new Bluebird((resolve, reject) => {
const chunks = chunk(batch, chunkSize);

Promise.resolve().then(async () => {
if (!isNumber(chunkSize) || chunkSize < 1) {
return reject(new TypeError(`Invalid chunkSize: ${chunkSize}`));
throw new TypeError(`Invalid chunkSize: ${chunkSize}`);
}

if (!Array.isArray(batch)) {
return reject(
new TypeError(`Invalid batch: Expected array, got ${typeof batch}`)
throw new TypeError(
`Invalid batch: Expected array, got ${typeof batch}`
);
}

//Next tick to ensure wrapper functions are called if needed
return Bluebird.delay(1)
.then(getTransaction)
.then((tr) => {
return Bluebird.mapSeries(chunks, (items) =>
tr(tableName).insert(items, returning)
)
.then((result) => {
result = flatten(result || []);
const chunks = chunk(batch, chunkSize);

if (autoTransaction) {
//TODO: -- Oracle tr.commit() does not return a 'thenable' !? Ugly hack for now.
return (tr.commit(result) || Bluebird.resolve()).then(
() => result
);
}
//Next tick to ensure wrapper functions are called if needed
await delay(1);
const tr = await getTransaction();

return result;
})
.catch((error) => {
if (autoTransaction) {
return tr.rollback(error).then(() => Bluebird.reject(error));
}
try {
const chunksResults = [];
for (const items of chunks) {
chunksResults.push(await tr(tableName).insert(items, returning));
}
const result = flatten(chunksResults);
if (autoTransaction) {
await tr.commit();
}
return result;
} catch (error) {
if (autoTransaction) {
await tr.rollback(error);
}

return Bluebird.reject(error);
});
})
.then(resolve)
.catch(reject);
throw error;
}
}),
{
returning(columns) {
Expand Down
3 changes: 3 additions & 0 deletions lib/util/delay.js
@@ -0,0 +1,3 @@
const { promisify } = require('util');

module.exports = promisify(setTimeout);

0 comments on commit 6de2633

Please sign in to comment.