socketmq - npm Package Compare versions

Comparing version 0.9.0 to 0.9.1

lib/message/wire.js

@@ -31,16 +31,15 @@ var amp = require('amp')
     // event name
-    event: args[1].toString(),
-    // message format id
-    formatId: args[2],
-    // message data
-    msg: args[3]
+    event: args[1].toString()
   }
   // message meta data
-  var metaIdx = 4
-
-  // message format id in string
-  if (pack.msg) {
+  var metaIdx
+  if (args.length > 3) {
+    metaIdx = 4
+    // message format id
+    pack.formatId = args[2]
     pack.fid = pack.formatId.toString()
+    // message data
+    pack.msg = args[3]
+    pack.hasMsg = true
   } else {
-    pack.formatId = undefined
     metaIdx = 2
@@ -47,0 +46,0 @@ }

lib/queue/channel.js

@@ -108,2 +108,5 @@ var type = require('../message/type')
       pack.msg = [chn]
+
+    // Set hasMsg to true so the dispatch function will use `msg` as arguments for callbacks.
+    pack.hasMsg = true
   }
@@ -110,0 +113,0 @@

lib/queue/hub.js

@@ -78,16 +78,18 @@ var type = require('../message/type')
       var tags
-      if (msg[REP]) {
-        tags = this.socket.prefixTags(REP, msg[REP])
-        if (meta && meta[MNS])
-          // We have namespace, tag again
-          tags = this.socket.prefixTags(meta[MNS], tags)
-        this.socket.tag(stream, tags)
+      if (msg) {
+        if (msg[REP]) {
+          tags = this.socket.prefixTags(REP, msg[REP])
+          if (meta && meta[MNS])
+            // We have namespace, tag again
+            tags = this.socket.prefixTags(meta[MNS], tags)
+          this.socket.tag(stream, tags)
+        }
+        if (msg[SUB]) {
+          tags = this.socket.prefixTags(SUB, msg[SUB])
+          if (meta && meta[MNS])
+            // We have namespace, tag again
+            tags = this.socket.prefixTags(meta[MNS], tags)
+          this.socket.tag(stream, tags)
+        }
       }
-      if (msg[SUB]) {
-        tags = this.socket.prefixTags(SUB, msg[SUB])
-        if (meta && meta[MNS])
-          // We have namespace, tag again
-          tags = this.socket.prefixTags(meta[MNS], tags)
-        this.socket.tag(stream, tags)
-      }
       this._flush()
@@ -94,0 +96,0 @@ break

lib/queue/index.js

@@ -151,3 +151,7 @@ var inf = require('./inf')
   this.beforeDispatch(pack, stream, function(pack, stream) {
-    var msg = pack.msg
+    var msg
+    if (true === pack.hasMsg)
+      msg = pack.msg || [pack.msg]
+    else
+      msg = []
     var meta = pack.meta
@@ -174,3 +178,3 @@ var event = pack.event
               }
-            } else if ('string' !== typeof repMsg && !isBuffer(repMsg)) {
+            } else if (1 === len && 'string' !== typeof repMsg && !isBuffer(repMsg)) {
               repMsg = [repMsg]
@@ -177,0 +181,0 @@ }

lib/socket/channel.js

@@ -33,23 +33,22 @@ var que = require('./que')
   var args = que.getMsgAndCallback(arguments, 2)
-  this.queue.req(this.streams, {
+  var pack = {
     chn: chn,
-    event: event,
-    msg: args.msg
-  }, args.callback)
+    msg: args.msg,
+    event: event
+  }
+  this.queue.req(this.streams, pack, args.callback)
 }
 
 SocketChannel.prototype.pubChn = function(chn, event, msg) {
-  this.queue.pub(this.streams, {
-    chn: chn,
-    event: event,
-    msg: que.getMsg(arguments, 2)
-  })
+  var pack = que.getMsg(arguments, 2)
+  pack.chn = chn
+  pack.event = event
+  this.queue.pub(this.streams, pack)
 }
 
 SocketChannel.prototype.pubSid = function(sid, event, msg) {
-  this.queue.pub(this.streams, {
-    sid: sid,
-    event: event,
-    msg: que.getMsg(arguments, 2)
-  })
+  var pack = que.getMsg(arguments, 2)
+  pack.sid = sid
+  pack.event = event
+  this.queue.pub(this.streams, pack)
 }
@@ -56,0 +55,0 @@

lib/socket/que.js

@@ -11,16 +11,13 @@ var Buffer = require('../message/type').Buffer
 exports.pub = function(event, msg) {
-  this.queue.pub(this.streams, {
-    event: event,
-    msg: getMsg(arguments, 1)
-  })
+  var pack = getMsg(arguments, 1)
+  pack.event = event
+  this.queue.pub(this.streams, pack)
 }
 
 exports.pubTag = function(tag, event, msg) {
-  msg = getMsg(arguments, 2)
+  var pack = getMsg(arguments, 2)
+  pack.event = event
   var streams = this.getStreamsByTag(tag)
   if (streams.length > 0) {
-    this.queue.pub(streams, {
-      event: event,
-      msg: msg
-    })
+    this.queue.pub(streams, pack)
   } else {
@@ -44,6 +41,7 @@ this.emit('error', {
   var args = getMsgAndCallback(arguments, 1)
-  this.queue.req(this.streams, {
-    event: event,
-    msg: args.msg
-  }, args.callback)
+  var pack = {
+    msg: args.msg,
+    event: event
+  }
+  this.queue.req(this.streams, pack, args.callback)
 }
@@ -53,8 +51,9 @@
   var args = getMsgAndCallback(arguments, 2)
+  var pack = {
+    msg: args.msg,
+    event: event
+  }
   var streams = this.getStreamsByTag(tag)
   if (streams.length > 0) {
-    this.queue.req(streams, {
-      event: event,
-      msg: args.msg
-    }, args.callback)
+    this.queue.req(streams, pack, args.callback)
   } else {
@@ -84,6 +83,10 @@ this.emit('error', {
 
-  return {
-    msg: getMsg(args, start, len),
-    callback: callback
-  }
+  var result
+  if (len > start)
+    result = getMsg(args, start, len)
+  else
+    result = {}
+
+  result.callback = callback
+  return result
 }
@@ -104,8 +107,5 @@
     }
-  } else {
-    // only one or no rgument is msg
+  } else if (end > start) {
+    // One msg arguments
     msg = args[start]
-    if (!msg)
-      return msg
-
     if ('string' !== typeof msg && !isBuffer(msg))
@@ -115,3 +115,5 @@ msg = [msg]
 
-  return msg
+  return {
+    msg: msg
+  }
 }

package.json

 {
   "name": "socketmq",
-  "version": "0.9.0",
+  "version": "0.9.1",
   "description": "Lightweight stream-oriented messaging library for node.",
@@ -14,9 +14,9 @@ "main": "index.js",
   "devDependencies": {
-    "coveralls": "2.11.14",
-    "engine.io": "1.7.2",
-    "engine.io-client": "1.7.2",
+    "coveralls": "2.11.15",
+    "engine.io": "2.0.1",
+    "engine.io-client": "2.0.1",
     "humanize-number": "0.0.2",
     "istanbul": "0.4.5",
     "nodemon": "1.11.0",
-    "tape": "4.6.2"
+    "tape": "4.6.3"
   },
@@ -23,0 +23,0 @@ "scripts": {

test/default.js

@@ -122,3 +122,3 @@ var test = require('tape')
   test(name + ': req/rep', function(t) {
-    t.plan(15)
+    t.plan(20)
 
@@ -168,2 +168,23 @@ // Single argument with callback
     smqClient2.req('without arguments')
+
+    // With only callback
+    var onlyCallbackMsg = 'onlyCallbackMsg'
+    smqServer.rep('with only callback', function(reply) {
+      t.equal(typeof reply, 'function', 'got reply function')
+      reply(onlyCallbackMsg)
+    })
+    smqClient2.req('with only callback', function(msg) {
+      t.equal(msg, onlyCallbackMsg, 'replied message match')
+    })
+
+    // With undefined message and callback
+    var undefinedCallbackMsg = 'undefinedCallbackMsg'
+    smqServer.rep('with undefined & callback', function(msg, reply) {
+      t.equal(msg, null, 'got msg value undefined')
+      t.equal(typeof reply, 'function', 'got reply function')
+      reply(undefinedCallbackMsg)
+    })
+    smqClient2.req('with undefined & callback', undefined, function(msg) {
+      t.equal(msg, undefinedCallbackMsg, 'replied message match')
+    })
   })
@@ -170,0 +191,0 @@

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

91500

increased by1.31%

Lines of code

2672

increased by1.06%

No dependency changes