Merge pull request #3 from jgn-epp/feature/peeking

Feature/peeking

Author: joelmenezes

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

@@ -1,5 +1,5 @@
-import {EventEmitter} from 'events';
-import {queueProxy} from './proxy';
+import { EventEmitter } from 'events';
+import { queueProxy } from './proxy';
 
 export default class Queue extends EventEmitter {
 
@@ -27,10 +27,10 @@ export default class Queue extends EventEmitter {
 		return new Queue(path);
 	}
 
-  static connectToRemoteQueue(path) {
-    queueProxy.connectRemote(path, true);
-    return new Queue(path);
-  }
+	static connectToRemoteQueue(path) {
+		queueProxy.connectRemote(path, true);
+		return new Queue(path);
+	}
 
 	startReceiving() {
 		if (this.receiving) {
@@ -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.