Version 1.0.68

lib/logging.js

@@ -6,6 +6,8 @@ var https = require("https");
 var moment = require("moment");
 var uuid = require("uuid");
 
+const accessor = {};
+
 module.exports = function (id, opts) {
 
 	opts = Object.assign({
@@ -24,7 +26,7 @@ module.exports = function (id, opts) {
 				keepAlive: true
 			})
 		},
-		credentials: configure.credentials
+		credentials: opts.credentials
 	});
 
 	var configFile = path.resolve(os.tmpdir(), `leolog_${id.toString()}.json`);
@@ -125,14 +127,15 @@ module.exports = function (id, opts) {
 						config.sequenceNumber = data.nextSequenceToken;
 						fs.writeFile(configFile, JSON.stringify(config, null, 2), callback);
 					}
-				})
+				});
 				logs = [];
 			}
 
 		});
 	}
 
 	var logger = {
+		sendEvents,
 		end: function (callback) {
 			process.stdout.write = oldStdOut;
 			process.stderr.write = oldStdErr;
@@ -149,7 +152,7 @@ module.exports = function (id, opts) {
 	process.once("beforeExit", () => {
 		logger.end((err) => {
 			if (err) {
-				console.log("Error uploading logs to aws:", err)
+				console.log("Error uploading logs to aws:", err);
 			}
 			console.log("Finished uploading logs", config.logGroupName);
 		});
@@ -159,6 +162,8 @@ module.exports = function (id, opts) {
 		// By catching and logging the error it adds to the event loop and causes "beforeExit" to fire
 		console.error(err);
 	});
-
+	accessor.logger = logger;
 	return logger;
-};
+};
+
+module.exports.accessor = accessor;

lib/stream/leo-stream.js

@@ -256,14 +256,30 @@ module.exports = function(configure) {
 				size: 1,
 				map: (payload, meta, done) => done(null, payload)
 			};
+
+			if (typeof opts.size != "object" && (opts.count || opts.records || opts.units || opts.time || opts.bytes)) {
+				opts.size = {};
+				opts.size.count = opts.count || opts.records || opts.units;
+				opts.size.time = opts.time;
+				opts.size.bytes = opts.size || opts.bytes;
+				opts.size.highWaterMark = opts.highWaterMark || 2;
+
+			}
+
 			if (!opts.batch || typeof opts.batch === "number") {
 				batch.size = opts.batch || batch.size;
 			} else {
 				batch.size = opts.batch.size || batch.size;
 				batch.map = opts.batch.map || batch.map;
 			}
+			if (typeof batch.size != "object") {
+				batch.size = {
+					count: batch.size
+				}
+			}
+			batch.size.highWaterMark = batch.size.highWaterMark || 2;
 
-			var batchSize = typeof batch.size === "number" ? batch.size : batch.size.count;
+			var batchSize = typeof batch.size === "number" ? batch.size : (batch.size.count || batch.size.records);
 			return ls.pipe(
 				ls.fromLeo(id, inQueue, opts),
 				ls.through((obj, done) => {
@@ -278,7 +294,9 @@ module.exports = function(configure) {
 					});
 				}),
 				ls.batch(batch.size),
-				ls.through((batch, done) => {
+				ls.through({
+					highWaterMark: 1
+				}, (batch, done) => {
 					batch.units = batch.payload.length;
 					let last = batch.payload[batch.units - 1];
 					if (batchSize == 1) {
@@ -290,12 +308,14 @@ module.exports = function(configure) {
 						done(null, batch);
 					}
 				}),
-				ls.process(id, func, null),
+				ls.process(id, func, null, undefined, {
+					highWaterMark: 1
+				}),
 				ls.toCheckpoint({
 					debug: opts.debug
 				}), callback);
 		},
-		process: function(id, func, outQueue, onflush) {
+		process: function(id, func, outQueue, onflush, opts = {}) {
 			var firstEvent;
 			var lastEvent;
 			var units;
@@ -343,7 +363,9 @@ module.exports = function(configure) {
 			}
 			reset();
 
-			return ls.through(function(obj, done) {
+			return ls.through({
+				highWaterMark: opts.highWaterMark
+			}, function(obj, done) {
 				if (!firstEvent) {
 					firstEvent = obj;
 				}
@@ -803,7 +825,6 @@ module.exports = function(configure) {
 			stream.on("end", function() {
 				console.log("end", arguments);
 				stream.checkpoint(opts || {}, () => {});
-				process.exit();
 			});
 
 			return stream;
@@ -923,7 +944,7 @@ module.exports = function(configure) {
 
 			logger.info(opts);
 			opts = Object.assign({
-				buffer: 1000,
+				buffer: 16,
 				loops: 100,
 				start: null,
 				limit: Number.POSITIVE_INFINITY,
@@ -1345,7 +1366,7 @@ module.exports = function(configure) {
 													logger.error(err);
 													done(err);
 												} else {
-													logger.debug("gzipped event read finished");
+													logger.debug("gzipped event read finished", item.records, totalCount);
 													done();
 												}
 											});

lib/streams.js

@@ -181,7 +181,7 @@ let ls = module.exports = {
 						stream.emit("error", err);
 					}
 					if (done) {
-						done(err);
+								done(err);
 					}
 				}, data);
 			} else {
@@ -199,7 +199,8 @@ let ls = module.exports = {
 					//want to add the flush statistics and send it further
 					done(null, obj);
 				});
-			}
+			},
+			highWaterMark: opts.highWaterMark || undefined
 		}, opts.commands), function(o, done) {
 			each.call(stream, o, (err, obj) => {
 				if (obj) {
@@ -579,7 +580,8 @@ let ls = module.exports = {
 		opts = Object.assign({
 			count: undefined,
 			bytes: undefined,
-			time: undefined
+			time: undefined,
+			highWaterMark: 2
 		}, opts);
 
 		var buffer = [];
@@ -617,6 +619,7 @@ let ls = module.exports = {
 			size: opts.size || opts.bytes,
 			records: opts.records || opts.count,
 			buffer: opts.buffer,
+			highWaterMark: opts.highWaterMark,
 			debug: opts.debug
 		}, function(obj, callback) {
 			let size = 0;
@@ -636,7 +639,7 @@ let ls = module.exports = {
 			});
 		}, function emit(callback, data) {
 			push(stream, data);
-			callback();
+			callback();	
 		}, function flush(callback) {
 			logger.debug("Batch On Flush");
 			callback();

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "leo-sdk",
-	"version": "1.0.68",
+	"version": "1.0.71",
 	"description": "Load data onto the LEO Platform",
 	"homepage": "https://leoplatform.io",
 	"main": "index.js",