electrodb - npm Package Compare versions

Comparing version 0.9.49 to 0.9.50

package.json

 {
   "name": "electrodb",
-  "version": "0.9.49",
+  "version": "0.9.50",
   "description": "A library to more easily create and interact with multiple entities and heretical relationships in dynamodb",
@@ -5,0 +5,0 @@ "main": "index.js",

src/entity.js

@@ -163,3 +163,4 @@ "use strict";
 			lastEvaluatedKeyRaw: !!options.lastEvaluatedKeyRaw,
-			table: options.table
+			table: options.table,
+			concurrent: options.concurrent
 		};
@@ -205,7 +206,15 @@ let stackTrace = new e.ElectroError(e.ErrorCodes.AWSError);
 		let results = [];
-		await Promise.all(parameters.map(async params => {
-			let response = await this._exec(MethodTypes.batchWrite, params);
-			let unprocessed = this.formatBulkWriteResponse(params.IndexName, response, config);
-			results = [...results, ...unprocessed];
-		}));
+		let concurrent = this._normalizeConcurrencyValue(config.concurrent)
+		let concurrentOperations = utilities.batchItems(parameters, concurrent);
+		// console.log("PUT", JSON.stringify(concurrentOperations));
+		for (let operation of concurrentOperations) {
+			await Promise.all(operation.map(async params => {
+				let response = await this._exec(MethodTypes.batchWrite, params);
+				let unprocessed = this.formatBulkWriteResponse(params.IndexName, response, config);
+				for (let u of unprocessed) {
+					results.push(u);
+				}
+			}));
+		}
+
 		return results;
@@ -218,10 +227,19 @@ }
 		}
+		let concurrent = this._normalizeConcurrencyValue(config.concurrent)
+		let concurrentOperations = utilities.batchItems(parameters, concurrent);
+
 		let resultsAll = [];
 		let unprocessedAll = [];
-		await Promise.all(parameters.map(async params => {
-			let response = await this._exec(MethodTypes.batchGet, params);
-			let [results, unprocessed] = this.formatBulkGetResponse(params.IndexName, response, config);
-			resultsAll = [...resultsAll, ...results];
-			unprocessedAll = [...unprocessedAll, ...unprocessed];
-		}));
+		for (let operation of concurrentOperations) {
+			await Promise.all(operation.map(async params => {
+				let response = await this._exec(MethodTypes.batchGet, params);
+				let [results, unprocessed] = this.formatBulkGetResponse(params.IndexName, response, config);
+				for (let r of results) {
+					resultsAll.push(r);
+				}
+				for (let u of unprocessed) {
+					unprocessedAll.push(u);
+				}
+			}));
+		}
 		return [resultsAll, unprocessedAll];
@@ -284,3 +302,2 @@ }
 		}
-
 		if (response.UnprocessedKeys[table] && response.UnprocessedKeys[table].Keys && Array.isArray(response.UnprocessedKeys[table].Keys)) {
@@ -393,2 +410,10 @@ for (let value of response.UnprocessedKeys[table].Keys) {
 
+	_normalizeConcurrencyValue(value = 1) {
+		value = parseInt(value);
+		if (isNaN(value) || value < 1) {
+			throw new e.ElectroError(e.ErrorCodes.InvalidConcurrencyOption, "Query option 'concurrency' must be of type 'number' and greater than zero.");
+		}
+		return value;
+	}
+
 	_deconstructKeys(index, keyType, key, backupFacets = {}) {
@@ -395,0 +420,0 @@ if (typeof key !== "string" || key.length === 0) {

src/errors.js

@@ -124,2 +124,8 @@ // # Errors:
   },
+  InvalidConcurrencyOption: {
+    code: 2004,
+    section: "invalid-concurrency-option",
+    name: "InvalidConcurrencyOption",
+    sym: ErrorCode
+  },
   InvalidAttribute: {
@@ -126,0 +132,0 @@ code: 3001,

No alert changes

Improved metrics

Total package byte prevSize

233508

increased by2.64%

Lines of code

3732

increased by0.78%

Number of lines in readme file

3056

increased by1.66%

No dependency changes