smqp - npm Package Compare versions

Comparing version 0.6.1 to 0.6.2

dist/index.js

@@ -71,7 +71,7 @@ 'use strict';
 
-  function subscribeOnce(exchangeName, routingKey, onMessage) {
+  function subscribeOnce(exchangeName, routingKey, onMessage, options = {}) {
     if (typeof onMessage !== 'function') throw new Error('message callback is required');
 
     const onceQueueName = generateId();
-    const onceConsumer = subscribe(exchangeName, routingKey, onceQueueName, wrappedOnMessage, { durable: false, noAck: true });
+    const onceConsumer = subscribe(exchangeName, routingKey, onceQueueName, wrappedOnMessage, { durable: false, noAck: true, consumerTag: options.consumerTag });
     return onceConsumer;
@@ -574,3 +574,3 @@
 
-      const consumerMessages = messages.filter(message => message.consumerTag === consumer.options.consumerTag);
+      const consumerMessages = messages.filter(message => message.consumerTag === consumer.consumerTag);
       for (let i = 0; i < consumerMessages.length; i++) {
@@ -719,4 +719,9 @@ consumerMessages[i].unsetConsumer();
   function Consumer(queueName, onMessage, options) {
-    const consumerOptions = Object.assign({ consumerTag: generateId(), prefetch: 1, priority: 0 }, options);
+    const consumerOptions = Object.assign({ prefetch: 1, priority: 0 }, options);
+    if (!consumerOptions.consumerTag) consumerOptions.consumerTag = generateId();else if (consumers.find(c => c.consumerTag === consumerOptions.consumerTag)) {
+      throw new Error(`Consumer tag must be unique, ${consumerOptions.consumerTag} is occupied`);
+    }
+
     const { consumerTag, noAck, priority } = consumerOptions;
+
     let prefetch;
@@ -723,0 +728,0 @@ setPrefetch(consumerOptions.prefetch);

index.js

@@ -65,7 +65,7 @@ export function Broker(source) {
 
-  function subscribeOnce(exchangeName, routingKey, onMessage) {
+  function subscribeOnce(exchangeName, routingKey, onMessage, options = {}) {
     if (typeof onMessage !== 'function') throw new Error('message callback is required');
 
     const onceQueueName = generateId();
-    const onceConsumer = subscribe(exchangeName, routingKey, onceQueueName, wrappedOnMessage, {durable: false, noAck: true});
+    const onceConsumer = subscribe(exchangeName, routingKey, onceQueueName, wrappedOnMessage, {durable: false, noAck: true, consumerTag: options.consumerTag});
     return onceConsumer;
@@ -570,3 +570,3 @@
 
-      const consumerMessages = messages.filter((message) => message.consumerTag === consumer.options.consumerTag);
+      const consumerMessages = messages.filter((message) => message.consumerTag === consumer.consumerTag);
       for (let i = 0; i < consumerMessages.length; i++) {
@@ -715,4 +715,10 @@ consumerMessages[i].unsetConsumer();
   function Consumer(queueName, onMessage, options) {
-    const consumerOptions = Object.assign({consumerTag: generateId(), prefetch: 1, priority: 0}, options);
+    const consumerOptions = Object.assign({prefetch: 1, priority: 0}, options);
+    if (!consumerOptions.consumerTag) consumerOptions.consumerTag = generateId();
+    else if (consumers.find((c) => c.consumerTag === consumerOptions.consumerTag)) {
+      throw new Error(`Consumer tag must be unique, ${consumerOptions.consumerTag} is occupied`);
+    }
+
     const {consumerTag, noAck, priority} = consumerOptions;
+
     let prefetch;
@@ -719,0 +725,0 @@ setPrefetch(consumerOptions.prefetch);

package.json

 {
   "name": "smqp",
-  "version": "0.6.1",
+  "version": "0.6.2",
   "description": "Synchronous message queuing package",
@@ -5,0 +5,0 @@ "author": {

README.md

 SMQP
 ====
 
-[![Build Status](https://travis-ci.org/paed01/smqp.svg?branch=master)](https://travis-ci.org/paed01/smqp)[![Build status](https://ci.appveyor.com/api/projects/status/8dy3yrde5pe8mk6m/branch/master?svg=true)](https://ci.appveyor.com/project/paed01/smqp/branch/master)[![Project Status: WIP – Initial development is in progress, but there has not yet been a stable, usable release suitable for the public.](http://www.repostatus.org/badges/latest/wip.svg)](http://www.repostatus.org/#wip)
+[![Build Status](https://travis-ci.org/paed01/smqp.svg?branch=master)](https://travis-ci.org/paed01/smqp)[![Build status](https://ci.appveyor.com/api/projects/status/8dy3yrde5pe8mk6m/branch/master?svg=true)](https://ci.appveyor.com/project/paed01/smqp/branch/master)[![Project Status: Active – The project has reached a stable, usable state and is being actively developed.](http://www.repostatus.org/badges/latest/active.svg)](http://www.repostatus.org/#active)
 
@@ -6,0 +6,0 @@ Synchronous message queuing package. Used as an alternative - and frontend ready - event handler when you expect events to be handled in sequence.

No alert changes

Improved metrics

Total package byte prevSize

50238

increased by1.09%

Lines of code

1448

increased by0.49%

No dependency changes