Feature/faster stream parsing

.github/workflows/node.js.yml

@@ -16,13 +16,13 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [12.x, 14.x]
+        node-version: [14.x, 16.x]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - name: CI/CD ${{ matrix.node-version }}
-      uses: actions/setup-node@v2
+      uses: actions/setup-node@v3
       with:
         node-version: ${{ matrix.node-version }}
         cache: 'npm'

.github/workflows/npm-publish.yml

@@ -14,7 +14,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-node@v3
         with:
-          node-version: 12
+          node-version: 16
           registry-url: https://registry.npmjs.org/
       - run: npm ci
       - run: npm i aws-sdk@^2.581.0

.gitignore

@@ -32,3 +32,5 @@ package-lock.json
 *.utest.js
 docs/build/
 /docs/build-docs.js
+scripts/*
+util-lambdas/*

index.d.ts

@@ -200,6 +200,18 @@ export declare class RStreamsSdk {
 	putEvent: <T>(bot_id: string, outQueue: string, payload: Event<T> | T) => Promise<void>;
 
 
+	/**
+	 * An async/await friendly function to write an array of events to a queue.
+	 *
+	 * @typeParam T The data to write as the payload of the event
+	 * @param payloads The payloads of the events to write
+	 * @param settings The botId and queue to use if payloas is a T[] instead of Event<T>[], and writeOptions for sdk.load()
+	 * @method
+	 * @todo example
+	 */
+	putEvents: <T>(payloads: (Event<T> | T)[], settings?: { botId?: string, queue?: string, writeOptions?: WriteOptions }) => Promise<void>;
+
+
 	/** @method */
 	throughAsync: typeof StreamUtil.throughAsync;
 

index.js

@@ -179,6 +179,12 @@ function SDK(id, data, awsResourceConfig) {
 		putEvent: function(bot_id, queue, payload) {
 			return promisify(this.put).call(this, bot_id, queue, payload);
 		},
+		putEvents: function(payloads, settings = {}) {
+			return this.streams.pipeAsync(
+				this.streams.eventstream.readArray(payloads),
+				this.load(settings.botId, settings.queue, settings.writeOptions)
+			);
+		},
 		throughAsync: leoStream.throughAsync,
 		checkpoint: leoStream.toCheckpoint,
 		streams: leoStream,

lib/aws-sdk-sync.ts

@@ -6,8 +6,7 @@ import { spawnSync } from "child_process";
 // use `module.require` to keep webpack from overriding the function.
 // This isn't run within the bundle
 // It is stringified and run in a different process
-export function invoke(service: string, method: string, config: any, params: any) {
-	let AWS = module.require.call(module, "aws-sdk");
+export function invoke(AWS: any, service: string, method: string, config: any, params: any) {
 	let hasLogged = false;
 	try {
 		new AWS[service](config)[method](params, (err: any, data: any) => {
@@ -30,7 +29,7 @@ export function invoke(service: string, method: string, config: any, params: any
 }
 
 function run(service: string, method: string, config: any, params: any) {
-	let fn = `(${invoke.toString()})("${service}", "${method}", ${JSON.stringify(config)}, ${JSON.stringify(params)})`;
+	let fn = `(${invoke.toString()})(require("aws-sdk"),"${service}", "${method}", ${JSON.stringify(config)}, ${JSON.stringify(params)})`;
 
 	// Spawn node with the function to run `node -e (()=>{})`
 	// Using `RESPONSE::{}::RESPONSE` to denote the response in the output
@@ -75,6 +74,13 @@ export class S3 extends Service<AWS.S3.ClientConfiguration> {
 	}
 }
 
+
+export class DynamoDB extends Service<AWS.DynamoDB.ClientConfiguration> {
+	putItem(): AWS.DynamoDB.PutItemOutput {
+		return this.invoke("putItem");
+	}
+}
+
 export default {
 	SecretsManager,
 	S3

lib/aws-util.ts

@@ -32,7 +32,9 @@ export function error(err, options) {
 	if (typeof err.message === 'string' && err.message !== '') {
 		if (typeof options === 'string' || (options && options.message)) {
 			originalError = copy(err);
-			originalError.message = err.message;
+			if (originalError != null) {
+				originalError.message = err.message;
+			}
 		}
 	}
 	err.message = err.message || null;

lib/configuration-builder.ts

@@ -74,6 +74,12 @@ export class ConfigurationBuilder<T> {
 			g.rstreams_project_config_cache = {};
 		}
 
+		if (g.rsf_config_opts) {
+			Object.assign(options, g.rsf_config_opts);
+		} else if (process.env.RSF_CONFIG_OPTS) {
+			Object.assign(options, JSON.parse(process.env.RSF_CONFIG_OPTS));
+		}
+
 		if (this.data == null || this.data == "") {
 			if (process.env.RSF_CONFIG) {
 				this.data = process.env.RSF_CONFIG;
@@ -101,15 +107,15 @@ export class ConfigurationBuilder<T> {
 					region: options.region
 				}).getSecretValue({
 					SecretId: this.data
-				}).SecretString);
+				}).SecretString.replace(/"{/g, '{').replace(/}"/g, "}"));
 				logger.timeEnd("get-rsf-config");
 			}
 		}
 
 		// Allow extra env vars to be defined as RSF_CONFIG_some.new.field=my_value
 		Object.entries(process.env).forEach(([key, value]) => {
 			const a = (key.match(/^RSF_CONFIG_(.*)$/) || [])[1];
-			if (a) {
+			if (a && key != "RSF_CONFIG_OPTS") {
 				let parts = a.split(".");
 				let lastPart = parts.pop();
 				let parent = parts.reduce((a, b) => {
@@ -143,10 +149,11 @@ export class ConfigurationBuilder<T> {
 		let returnValue = {};
 		Object.getOwnPropertyNames(root).forEach(key => {
 			let value = root[key];
+			let origKey = key;
 
 			// convert  string shorthand to full ResourceReference
 			if (typeof value === "string" && value.match(/^.+?::/)) {
-				let [service, key, type, opts] = value.split('::');
+				let [service, key, type, opts] = value.split(/(?<!AWS)::/);
 				type = type || "dynamic";
 				value = {
 					service,
@@ -160,6 +167,11 @@ export class ConfigurationBuilder<T> {
 						return all;
 					}, {}))
 				};
+
+				// If it isn't a valid reference, set it back 
+				if (!this.isResourceReference(value)) {
+					value = root[origKey];
+				}
 			}
 
 			if (this.isResourceReference(value)) {
@@ -211,10 +223,10 @@ export class ConfigurationBuilder<T> {
 		}
 	}
 	private isResourceReference(value: any): boolean {
-		return value != null && typeof value === "object" && value.service && value.key && value.type && ConfigurationBuilder.Resolvers[value.service];
+		return value != null && typeof value === "object" && value.service && value.key && value.type && ConfigurationBuilder.Resolvers[value.service] != null;
 	}
 
-	static Resolvers = {
+	static Resolvers: Record<string, (ref: ResourceReference, cache: any) => any> = {
 		// ssm: (ref: ResourceReference) => {
 		// 	return process.env[`RS_ssm::${resolveKeywords(ref.key, ref.options)}`];
 		// },
@@ -256,7 +268,7 @@ export class ConfigurationBuilder<T> {
 	};
 }
 
-function resolveKeywords(template: string, data: any) {
+export function resolveKeywords(template: string, data: any) {
 	const name = template.replace(/\${(.*?)}/g, function (match, field) {
 		let value = getDataSafe(data, field);
 		if (value != null && typeof value === "object") {
@@ -266,7 +278,7 @@ function resolveKeywords(template: string, data: any) {
 	}).replace(/[_-]{2,}/g, "");
 	return name;
 }
-function getDataSafe(data = {}, path = "") {
+export function getDataSafe(data = {}, path = "") {
 	const pathArray = path.split(".").filter(a => a !== "");
 	if (pathArray.length === 0) {
 		return data;
@@ -281,7 +293,7 @@ const nullRegex = /^null$/;
 const undefinedRegex = /^undefined$/;
 const jsonRegex = /^{(.|\n)*}$/;
 
-function inferTypes(node) {
+export function inferTypes(node) {
 	let type = typeof node;
 	if (Array.isArray(node)) {
 		for (let i = 0; i < node.length; i++) {