vubsub - npm Package Compare versions

Comparing version 1.0.5 to 1.0.6

lib/channel.js

@@ -40,3 +40,3 @@ var util = require('util')
         tailable: true,
-        awaitdata: true,
+        //awaitdata: true,
         numberOfRetries: -1,
@@ -53,3 +53,4 @@ tailableRetryInterval: Channel.TAILABLE_RETRY_INTERVAL
     });
-  });   
+  });
+  self.db.collection('channels').insert({ client: self.client.id, channel: self.name, ns: self.client.namespace, ping: new Date() });
 }
@@ -60,33 +61,33 @@ util.inherits(Channel, EventEmitter);
 
-Channel.collection = function(db, namespace, cb) {
-  var d = q.defer();
-  db.createCollection('ns_' + namespace, {
-    size: Channel.COLLECTION_SIZE,
-    capped: true,
-    autoIndexId: true
-  }, function(err, collection) {
-    if (err) {
-      d.reject(err);
-    } else if (!collection) {
-      d.reject(new Error('failed to create a new namespace collection'));
-    } else {
-      d.resolve(collection);
-    }
-  });
-  return d.promise.nodeify(cb);
+Channel.prototype.send = function(type, data, cb) {
+  var self = this;
+  self.ping();
+  return self.collection.then(function(collection) {
+    return q.ninvoke(collection, 'insert', {
+      ts: new Date(),
+      channel: self.name,
+      from: self.client.id,
+      type: type,
+      data: data
+    });
+  }).nodeify(cb);
 }
-Channel.send = function(db, namespace, message, cb) {
-  return Channel.collection(db, namespace).then(function(collection) {
-    return q.ninvoke(collection, 'insert', message);    
+Channel.prototype.leave = function(cb) {
+  return this.client.leave(this, cb);
+}
+Channel.prototype.ping = function(cb) {
+  var self = this;
+  return self.client.ping().then(function() {
+    return q.nfcall(self.db.collection('channels').update, { client: self.client.id, channel: self.name }, { $set: { ping: new Date() }});
   }).nodeify(cb);
 }
-
+Channel.prototype.count = function(cb) {
+  return Channel.count(this.db, this.client.namespace, this.name, cb);
+}
 Channel.prototype._handleNext = function(err, data) {
   var self = this;
   if (err || !data) {
-console.error('Channel._handleNext', self.name, err, data);
     self.cursor.close();
     self.emit('error', err ? err : new Error('no data'));
   } else {
-console.log('Channel._handleNext', self.name, data);
     if (data.type) self.emit(data.type, data);
@@ -101,3 +102,3 @@
 }
-Channel.prototype._close = function() {
+Channel.prototype._close = function(cb) {
   if (this.cursor) {
@@ -107,20 +108,31 @@ this.cursor.close();
   }
+  return q.ninvoke(this.db.collection('channels'), 'remove', { client: this.client.id, channel: this.name }).nodeify(cb);
 }
 
-Channel.prototype.send = function(type, data, cb) {
-  var self = this;
-  return self.collection.then(function(collection) {
-    return q.ninvoke(collection, 'insert', {
-      ts: new Date(),
-      channel: self.name,
-      from: self.client.id,
-      type: type,
-      data: data
-    });
+Channel.collection = function(db, namespace, cb) {
+  var d = q.defer();
+  db.createCollection('ns_' + namespace, {
+    size: Channel.COLLECTION_SIZE,
+    capped: true,
+    autoIndexId: true
+  }, function(err, collection) {
+    if (err) {
+      d.reject(err);
+    } else if (!collection) {
+      d.reject(new Error('failed to create a new namespace collection'));
+    } else {
+      d.resolve(collection);
+    }
+  });
+  return d.promise.nodeify(cb);
+}
+Channel.send = function(db, namespace, message, cb) {
+  return Channel.collection(db, namespace).then(function(collection) {
+    return q.ninvoke(collection, 'insert', message);    
   }).nodeify(cb);
 }
-Channel.prototype.leave = function() {
-  self.client.leave(this);
+Channel.count = function(db, namespace, channel, cb) {
+  return q.ninvoke(db.collection('channels').find({ channel: channel, ns: namespace }), 'count').nodeify(cb);
 }
 
 module.exports = Channel;

lib/client.js

@@ -12,3 +12,3 @@ var q = require('q')
 Client.prototype.ping = function(cb) {
-  this.db.collection('clients').update({ _id: this.id }, { $set: { ping: new Date() }}, cb);
+  return q.nfcall(this.db.collection('clients').update, { _id: this.id }, { $set: { ping: new Date() }}).nodeify(cb);
 }
@@ -21,14 +21,8 @@ Client.prototype.channel = Client.prototype.join = function(name, lastMessage) {
     this.channels[name] = channel;
-    var doc = { $set: { } };
-    doc.$set["channels." + channel.name] = new Date();
-    this.db.collection('clients').update({ _id: this.id }, doc);
     return channel;
   }
 }
-Client.prototype.leave = function(channel) {
-  var doc = { $unset: { } };
-  doc.$unset["channels." + channel.name] = true;
-  this.db.collection('clients').update({ _id: this.id }, doc);
-  channel._close();
-  delete this.channels[name];
+Client.prototype.leave = function(channel, cb) {
+  delete this.channels[channel.name];
+  return channel._close(cb);
 }
@@ -43,3 +37,3 @@ Client.prototype.send = function(channel, type, data) {
 var clients = {};
-exports.create = function(db, meta) {
+exports.create = function(db, meta, cb) {
   var meta = meta || {};
@@ -60,5 +54,5 @@ meta.ns = meta.ns || 'vubsub';
   });
-  return deferred.promise;
+  return deferred.promise.nodeify(cb);
 }
-exports.get = function(id) {
+exports.get = function(id, cb) {
   return q.fcall(function() {
@@ -70,5 +64,6 @@ if (clients[id]) {
     }   
-  });
+  }).nodeify(cb);
 }
 exports.send = Channel.send;
+exports.count = Channel.count;
 

package.json

 {
   "name": "vubsub",
-  "version": "1.0.5",
+  "version": "1.0.6",
   "description": "Pub/Sub for Node.js and MongoDB",
@@ -5,0 +5,0 @@ "homepage": "https://github.com/vivocha/vubsub",

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

9975

increased by5.67%

Lines of code

190

increased by4.4%

No dependency changes