promise-stream-queue - npm Package Compare versions

Comparing version 0.2.2 to 0.3.0

examples/async.js

 const Stream = require('../main.js');
 
 var stream = new Stream(5000);
-var nums = [];
+var i=0;
 
@@ -11,12 +11,13 @@ // Iterate on the stream
 	else console.log("RESULT => Resolve\t",data);
+}).done(()=>{
+	console.log("Done!");
 });
 
-// Creates 100 asynchronous promises
-for(var i=0;i<100;i++) nums.push(i);
-nums.forEach(i=>{
+// Creates asynchronous promises
+setInterval(()=>{
+	var seq = i++;
 	stream.push(new Promise((resolve,reject)=>{
 		var rnd = Math.random();
 		var to = Math.floor(Math.random()*10000);
-		var data = "PR => "+i+", "+to;
-		console.log(data);
+		var data = "PR => "+seq+", "+to;
 		setTimeout(()=>{
@@ -27,2 +28,11 @@ if(rnd<=0.5) {resolve(data);}
 	}));
-});
+},100);
+
+// Close and drain stream after 10 seconds. No new promises
+// are pushed to the stream, and the pending ones are
+// cancelled
+setTimeout(()=>{
+	console.log("Closing stream...");
+	stream.close();
+	stream.drain();
+},10000);

main.js

@@ -15,3 +15,7 @@ const
 	var callbacks = [];
+	var dones = [];
+	var closed = false;
 
+	function voidfn(){};
+
 	function notify(event,data) {
@@ -27,2 +31,3 @@ buffer.pop();
 		var defs = [];
+		var last = closed && !buffer.length;
 		callbacks.forEach(cb=>{
@@ -59,2 +64,9 @@ if(!cb.sync) cb.fn(err,res,ex);
 		}
+		else {
+			if(closed) {
+				while(dones.length)
+					dones.pop()();
+				self.emit("done");
+			}
+		}
 	}
@@ -67,2 +79,7 @@
 	this.push = function(pr) {
+		if(closed) {
+			pr.then(voidfn,voidfn).catch(voidfn);
+			return;
+		}
+
 		id++;
@@ -84,4 +101,24 @@ var item = new Item("Item_"+id,pr,timeout);
 
+	this.close = function() {
+		closed = true;
+		self.emit("closed");
+	}
+
+	this.done = function(callback) {
+		dones.push(callback);
+
+		if(closed) {
+			while(dones.length)
+				dones.pop()();
+			self.emit("done");
+		}
+	}
+
+	this.drain = function() {
+		buffer.forEach(b=>b.kill("Drain"));
+	}
+
 	this.forEach = function(callback) {
 		callbacks.push({fn:callback,sync:false});
+		return this;
 	}
@@ -91,2 +128,3 @@
 		callbacks.push({fn:callback,sync:true});
+		return this;
 	}
@@ -93,0 +131,0 @@ }

package.json

 {
   "name": "promise-stream-queue",
-  "version": "0.2.2",
+  "version": "0.3.0",
   "description": "Promise Stream. Queue promises and retrieve the resolved/rejected ones in the inserted order",
@@ -5,0 +5,0 @@ "author": "David Gómez Matarrodona <[e-mail]>",

README.md

@@ -99,2 +99,11 @@ # promise-stream-queue
 
+### stream.close()
+Closes the stream. New pushed promises will be ignored, and
+the stream will be marked as closed. The pending promises
+are still being processed.
+
+### stream.drain()
+Drains the stream. Rejects all the pending promises of the
+queue.
+
 ### stream.toArray()
@@ -107,6 +116,8 @@ Returns a snapshot of the stream as an static array of promises.
 ### stream.forEachSync(callback(err,res,ex,next))
-Same as before, but now, the *next* argument is a function that must be called
-in order to move to the next element. This is useful when we want to wait to
-the callback function to finish the process before move to the next promise.
+Same as before, but now, the *next* argument is a function that must be called in order to move to the next element. This is useful when we want to wait to the callback function to finish the process before move to the next promise.
 
+### stream.done(callback)
+The callback function will be invoked when the stream is closed and empty. Useful when the stream has been closed and we want to
+know when all promises have been processed
+
 ## Events
@@ -131,4 +142,16 @@ ### resolve
 
+### closed
+Fired when the stream has been closed
+```javascript
+stream.on("closed",()=>console.log('Closed'));
+```
+
+### done
+Fired when the stream is closed and all promises has been processed
+```javascript
+stream.on("done",()=>console.log('Done'));
+```
+
 ## Examples
 * [Asynchronous stream iteration](https://github.com/solzimer/promise-stream-queue/blob/master/examples/async.js)
 * [Synchronous stream iteration](https://github.com/solzimer/promise-stream-queue/blob/master/examples/sync.js)

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

13520

increased by12.55%

Lines of code

252

increased by19.43%

Number of lines in readme file

155

increased by17.42%

No dependency changes