Add peeking

Added peeking for messages without removing them untill explicitly done.
This is great for situations where you want to be absolutely sure that
the message was parsed and dealt with properly removing being removed
from the queue.

Author: jgn-epp

MSMQLib/MSMQInterface.cs

@@ -99,6 +99,47 @@ public async Task<object> ReceiveMessages(dynamic input)
             }
         }
 
+        /// <summary>
+        /// Removes a message with the given id from the queue.
+        /// </summary>
+        /// <param name="input">The queue's path and message id.</param>
+        public async Task<object> ReceiveMessageById(dynamic input)
+        {
+            var path = (string)input.path;
+            var id = (string)input.id;
+
+            MessageQueue queue = new MessageQueue(path);
+            queue.Formatter = new BinaryMessageFormatter();
+            queue.MessageReadPropertyFilter.SetAll();
+
+            try
+            {
+                var msg = (MSMQMessage)queue.ReceiveById(id);
+                return true;
+            }
+            catch (InvalidOperationException) // Thrown if message with given id is not in queue.
+            {
+                return false;
+            }
+        }
+
+        /// <summary>
+        /// Peeks a message from a queue without removing it.
+        /// </summary>
+        /// <param name="path">The queue's path.</param>
+        public async Task<object> Peek(string path)
+        {
+            MessageQueue queue = new MessageQueue(path);
+            queue.Formatter = new BinaryMessageFormatter();
+            queue.MessageReadPropertyFilter.SetAll();
+
+            var msg = await Task.Factory.FromAsync<Message>(
+                queue.BeginPeek(),
+                queue.EndPeek);
+
+            return (MSMQMessage)msg;
+        }
+
         /// <summary>
         /// Gets all messages from a queue without removing them.
         /// </summary>

proxy.js

@@ -7,15 +7,17 @@ const baseOptions = {
 };
 
 function getMethod(methodName) {
-	return edge.func(Object.assign({}, baseOptions, {methodName}));
+	return edge.func(Object.assign({}, baseOptions, { methodName }));
 }
 
 export var queueProxy = {
 	exists: getMethod('ExistsQueue'),
 	create: getMethod('CreateQueue'),
 	send: getMethod('SendMessage'),
 	receive: getMethod('ReceiveMessages'),
+	peek: getMethod('Peek'),
+	remove: getMethod('ReceiveMessageById'),
 	list: getMethod('GetAllMessages'),
-  clear: getMethod('PurgeQueue'),
-  connectRemote: getMethod('ConnectRemote')
+	clear: getMethod('PurgeQueue'),
+	connectRemote: getMethod('ConnectRemote')
 };

queue.js

@@ -47,6 +47,49 @@ export default class Queue extends EventEmitter {
 		});
 	}
 
+	startPeeking() {
+		if (this.receiving) {
+			throw new Error('Already receiving messages from this queue');
+		}
+
+		this.receiving = true;
+
+		const start = () => {
+			this.peek().then(msg => {
+				this.emit('peek', {
+					msg,
+					next: async () => {
+						await this.remove(msg.id);
+						start();
+					}
+				})
+			})
+		};
+
+		start();
+	}
+
+	peek() {
+		return new Promise((resolve, reject) => {
+			queueProxy.peek(this.path, (error, msg) => {
+				if (error) reject(error);
+				else resolve(msg);
+			});
+		});
+	}
+
+	remove(id) {
+		return new Promise((resolve, reject) => {
+			queueProxy.remove({
+				path: this.path,
+				id
+			}, (error, result) => {
+				if (error) reject(error);
+				else resolve(result);
+			});
+		})
+	}
+
 	send(message, cb) {
 		let formattedMessage = JSON.stringify(message);
 

readme.md

@@ -9,7 +9,7 @@ NPM package is published as updated-node-msmq. https://www.npmjs.com/package/upd
 ## Differences from `node-msmq`
 
 * Support for Node.Js 6.x, 7.x, 8.x, 9.x, 10.x
-* Support to push objects to the queue instead of just strings. 
+* Support to push objects to the queue instead of just strings.
 * Support to send/receive messages to/from a queue on a **remote** machine.
 
 ## Install
@@ -51,6 +51,55 @@ queue.on('receive', (msg) => {
 queue.startReceiving();
 ```
 
+### Peek messages
+
+Start listeng for messages without removing them until you are ready to do so. Callbacks for the 'peek' event receive an object containing two properties:
+
+* `msg` The MSMQ Message.
+* `next()` Pops the message from the queue and start peeking for the next one.
+
+```js
+queue.on('peek', (event) => {
+  console.log(event.msg.body);
+
+  // Do some logic that might fail.
+  if (Math.random() > 0.5) {
+    throw Error('number is too high!');
+  }
+
+  // Remove the message from the queue and peek for next message.
+  event.next();
+})
+
+queue.startPeeking();
+```
+
+### Peek
+
+Promised based method to peek for a message in the queue without removing it. Resolves to a MSMQMessage.
+
+```js
+queue.peek().then(msg => console.log(msg.body));
+
+//or
+
+let msg = await queue.peek();
+```
+
+### Remove
+
+Promise based method to remove a message with given id from the queue. Resolves to `true` if message was removed and `false` if message didn't exist in the queue.
+
+```js
+queue.remove('12345')
+     .then(status => console.log(status ? 'Message was removed'
+                                        : 'Message was not in queue'));
+
+//or
+
+let status = await queue.remove('12345');
+```
+
 ### Get all messages
 
 Gets all messages without removing them from queue.
@@ -124,7 +173,7 @@ var queue = msmq.connectToRemoteQueue('.\\Private$\\MyAwesomeQueue');
 var messages = queue.getAllMessages();
 ```
 
-#### Note: 
+#### Note:
 * Creating a queue / Checking if a queue exists on a remote machine is currently not supported by MSMQ.
 * To communicate with a remote queue, MSMQ should be enabled in the sender's machine too. Also, in the _Security_ tab of the queue on the remote machine should have the appropriate permissions set for _Everyone_ and _ANONYMOUS LOGON_.
 * The queue should already be created on the remote machine.