socketmq - npm Package Compare versions

Comparing version 0.11.0 to 0.12.0

lib/socket/conn.js

@@ -14,3 +14,3 @@ var Url = require('url')
   } else {
-    this.close(stream, {
+    this.removeStream(stream, {
       data: buf,
@@ -86,2 +86,4 @@ type: this.ERR_UNWRITABLE,
   if (options.reconnect) {
+    var self = this
+    self.reconnect = true
     var _reconnect = makeReconnector(connect)
@@ -94,2 +96,5 @@ connect = function() {
           .on('error', function() {})
+          .on('connect', function(client) {
+            client.reconnector = this
+          })
           .on('reconnect', function(n, delay) {
@@ -102,2 +107,7 @@ // Simplely emit reconnect event through `smq` when it's a real
           })
+          .on('disconnect', function() {
+            if (self.reconnect === false) {
+              this.reconnect = false
+            }
+          })
           .connect()._connection
@@ -144,3 +154,3 @@ )
 
-exports.close = exports.removeStream = function(stream, error) {
+exports.removeStream = function(stream, error) {
   if (error) {
@@ -155,3 +165,3 @@ this.emit('error', error)
   }
-  stream.destroy && stream.destroy()
+  stream.destroy && stream.destroy(error)
 }

lib/transport/common.js

@@ -1,3 +0,1 @@
-
-
 exports.setupStream = function(smq, stream, callback, server) {
@@ -8,11 +6,14 @@ stream.setNoDelay && stream.setNoDelay(true)
   stream.once('close', function() {
-    smq.close(stream)
+    smq.removeStream(stream)
   })
 
   stream.on('error', function(err) {
-    smq.close(stream, {
-      type: smq.ERR_STREAM,
-      error: err,
-      stream: stream
-    })
+    // Only remove stream with original error. Avoid removing twice.
+    if (err && !err.stream) {
+      smq.removeStream(stream, {
+        type: smq.ERR_STREAM,
+        error: err,
+        stream: stream
+      })
+    }
   })
@@ -19,0 +20,0 @@

lib/transport/engine.io/bind.js

@@ -8,3 +8,2 @@ var http = require('htt' + 'p')
 
-
 module.exports = function bind(target, smq, options, callback) {
@@ -11,0 +10,0 @@ var httpServer

lib/transport/engine.io/common.js

@@ -16,7 +16,7 @@ var Duplex = require('stream').Duplex
   if (eioSocket.port) {
-    // client side eioSocket
+    // client side socket
     stream.remotePort = eioSocket.port
     stream.remoteAddress = eioSocket.hostname
   } else if (eioSocket.request) {
-    // server side connection eioSocket
+    // server side connection socket
     stream.id = getSocketID(eioSocket.id)
@@ -42,6 +42,6 @@ stream.socket = eioSocket
   var closing = false
-  var destroy = function() {
-    if (!closing) {
+  var destroy = function(err) {
+    if (!closing && eioSocket.readyState !== 'closed') {
       closing = true
-      eioSocket.close()
+      eioSocket.close(err)
     }
@@ -67,25 +67,23 @@ }
     // Might be an ArrayBuffer in browser
-    if (buf && !Buffer.isBuffer(buf) && 'undefined' !== typeof buf.byteLength)
+    if (buf && !Buffer.isBuffer(buf) && 'undefined' !== typeof buf.byteLength) {
       buf = new Buffer(buf)
+    }
     stream.push(buf)
   })
 
-  var onClose = function(error) {
-    eioSocket.removeListener('close', onClose)
+  eioSocket.once('close', function(reason, desc) {
     stream.push(null)
     stream.writable = false
-    stream.emit('close', error)
-    stream.destroy()
-  }
+    stream.emit('close')
+  })
 
-  eioSocket.on('close', onClose)
-
-  var onError = function(err) {
-    eioSocket.removeListener('error', onError)
+  eioSocket.once('error', function(err) {
+    if (err && err.description === 400) {
+      if (stream.reconnector) {
+        stream.reconnector.reconnect = false
+      }
+    }
     stream.emit('error', err)
-    onClose(err)
-  }
+  })
 
-  eioSocket.on('error', onError)
-
   return stream
@@ -92,0 +90,0 @@ }

lib/transport/engine.io/connect.js

@@ -8,3 +8,2 @@ var eio = require('engine.io-client')
 
-
 module.exports = function connect(target, smq, options, callback) {
@@ -32,10 +31,12 @@ options = getDefaultOptions(options)
 
-  if (true === https)
+  if (true === https) {
     protocol = 'wss:'
+  }
 
   var uri = protocol + '//' + target.hostname
-  if (target.port)
+  if (target.port) {
     uri = uri + ':' + target.port
+  }
 
   return uri
 }

package.json

 {
   "name": "socketmq",
-  "version": "0.11.0",
+  "version": "0.12.0",
   "description": "Lightweight messaging library for node & browser.",
@@ -5,0 +5,0 @@ "main": "index.js",

test/default.js

@@ -328,3 +328,3 @@ var test = require('tape')
     })
-    smqClient1.close(clientStream1)
+    smqClient1.removeStream(clientStream1)
   })
@@ -484,3 +484,3 @@
       setTimeout(function() {
-        smqClient1.close(smqClient1.streams[0])
+        smqClient1.removeStream(smqClient1.streams[0])
       }, 100)
@@ -487,0 +487,0 @@ })

test/gateway.test.js

@@ -253,3 +253,3 @@ var fs = require('fs')
       t.equal(chn, 'my room', 're-join chn')
-      tcpClient.close(tcpStream)
+      tcpClient.removeStream(tcpStream)
     })
@@ -266,3 +266,3 @@
         t.equal(chn, 'my room', 'reconnect join chn')
-        eioClient.close(eioClient.streams[0])
+        eioClient.removeStream(eioClient.streams[0])
       })
@@ -269,0 +269,0 @@

No alert changes

Improved metrics

Total package byte prevSize

95596

increased by0.47%

Lines of code

2885

increased by0.52%

No dependency changes