volos-cache-common - npm Package Compare versions

Comparing version 0.9.4 to 0.9.5

lib/cache-argo.js

@@ -27,12 +27,3 @@ /****************************************************************************
 var _ = require('underscore');
-var debug;
-var debugEnabled;
-if (process.env.NODE_DEBUG && /cache/.test(process.env.NODE_DEBUG)) {
-  debug = function(x) {
-    console.log('Quota: ' + x);
-  };
-  debugEnabled = true;
-} else {
-  debug = function() { };
-}
+var eventEmitter = new (require('events').EventEmitter)();
 
@@ -61,36 +52,30 @@ function CacheArgo(cache, options) {
       var req = env.request;
-      if (req.method === 'GET') {
-        var resp = env.response;
-        if (_.isFunction(id)) { id = id(req); }
-        var key = id ? id : req.url;
+      if (req.method !== 'GET') { return next(env); }
 
-        debug('Cache check');
-        self.internalCache.get(key, function (err, reply) {
-          if (err) { console.log('Cache error: ' + err); }
-          resp.setHeader('Cache-Control', "public, max-age=" + Math.floor(options.ttl / 1000) + ", must-revalidate");
-          if (reply) {
-            if (debugEnabled) { debug('cache hit: ' + key); }
-            var len = reply.readUInt8(0);
-            var contentType = reply.toString('utf8', 1, len + 1);
-            var content = reply.toString('utf8', len + 1);
-            resp.setHeader('Content-Type', contentType);
-            resp.body = content;
-            env.argo._routed = true; // bypass further pipeline processing
-            resp.from_cache = true; // avoid double caching
-          } else {
-            if (debugEnabled) { debug('cache miss: ' + key); }
-          }
+      var resp = env.response;
+      if (_.isFunction(id)) { id = id(req); }
+      var key = id ? id : req.url;
+      req._key = key;
+
+      var getSetCallback = function(err, reply, fromCache) {
+        if (err) { console.log('Cache error: ' + err); }
+
+        if (reply && fromCache) {
+          if (debugEnabled) { debug('cache hit: ' + key); }
+          var len = reply.readUInt8(0);
+          var contentType = reply.toString('utf8', 1, len + 1);
+          var content = reply.toString('utf8', len + 1);
+          resp.setHeader('Content-Type', contentType);
+          resp.body = content;
+          env.argo._routed = true; // bypass further pipeline processing
+        }
+      };
+
+      var populate = function(key, cb) {
+        if (debugEnabled) { debug('cache miss: ' + key); }
+
+        eventEmitter.once(key, function(buffer) {
+          cb(null, buffer);
         });
-      }
-      return next(env);
-    });
 
-    handle('response', function(env, next) {
-      var req = env.request;
-      var resp = env.response;
-      if (req.method === 'GET' && !resp.from_cache) { // avoid double caching
-        if (_.isFunction(id)) { id = id(req); }
-        var key = id ? id : req.url;
-
-        // replace end() to intercept the content returned to the client
         var end = resp.end;
@@ -105,15 +90,24 @@ resp.end = function(chunk, encoding) {
           buffer.write(chunk, contentType.length + 1);
-          self.internalCache.set(key, buffer, options, function(err) {
-            if (err) {
-              console.log('Cache error: ' + err);
-            } else if (debugEnabled) {
-              debug('Cached: ' + key);
-            }
-          });
+          eventEmitter.emit(key, buffer);
           resp.end(chunk, encoding);
         };
-      }
-      return next(env);
+      };
+
+      debug('Cache check');
+      resp.setHeader('Cache-Control', "public, max-age=" + Math.floor(options.ttl / 1000) + ", must-revalidate");
+      self.internalCache.getSet(key, populate, options, getSetCallback);
+      next(env);
     });
   };
 };
+
+var debug;
+var debugEnabled;
+if (process.env.NODE_DEBUG && /cache/.test(process.env.NODE_DEBUG)) {
+  debug = function(x) {
+    console.log('Quota: ' + x);
+  };
+  debugEnabled = true;
+} else {
+  debug = function() { };
+}

lib/cache-connect.js

@@ -47,76 +47,80 @@ /****************************************************************************
   return function(req, resp, next) {
+
+    if (req.method !== 'GET') { return next(); }
+
     if (_.isFunction(id)) { id = id(req); }
     var key = id ? id : req.originalUrl;
     debug('Cache check');
-    if (req.method === 'GET') {
-      self.internalCache.get(key, function(err, reply) {
-        if (err) { console.log('Cache error: ' + err); }
-        resp.setHeader('Cache-Control', "public, max-age=" + Math.floor(options.ttl / 1000) + ", must-revalidate");
-        if (reply) {
-          if (debugEnabled) { debug('cache hit: ' + key); }
-          var len = reply.readUInt8(0);
-          var contentType = reply.toString('utf8', 1, len + 1);
-          var content = reply.toString('utf8', len + 1);
-          if (contentType !== '') { resp.setHeader('Content-Type', contentType); }
-          resp._fromCache = true; // avoid double caching
-          return resp.end(content);
-        } else {
-          if (debugEnabled) {
-            debug('cache miss: ' + key);
-          }
 
-          var didWrite = false; // if multiple writes attempted, dumps cache
+    var getSetCallback = function(err, reply, fromCache) {
+      if (err) { return console.log('Cache error: ' + err); }
 
-          // replace write() to intercept the content sent to the client
-          resp._v_write = resp.write;
-          resp.write = function (chunk, encoding) {
-            resp._v_write(chunk, encoding);
-            if (chunk) {
-              if (didWrite) {
-                self.internalCache.delete(key);
-              } else {
-                didWrite = true;
-                var contentType = resp._headers['content-type'] || '';
-                cache(self, key, options, contentType, chunk);
-              }
-            }
-          };
+      if (reply && fromCache) {
+        if (debugEnabled) { debug('cache hit: ' + key); }
+        var len = reply.readUInt8(0);
+        var contentType = reply.toString('utf8', 1, len + 1);
+        var content = reply.toString('utf8', len + 1);
+        if (contentType !== '') {
+          resp.setHeader('Content-Type', contentType);
+        }
+        return resp.end(content);
+      }
+    };
 
-          // replace end() to intercept the content returned to the client
-          if (!didWrite) {
-            var end = resp.end;
-            resp.end = function (chunk, encoding) {
-              resp.end = end;
-              if (chunk && !resp._fromCache) { // avoid double caching
-                resp.on('finish', function () {
-                  var contentType = resp._headers['content-type'] || '';
-                  cache(self, key, options, contentType, chunk);
-                });
-              }
-              resp.end(chunk, encoding);
-            };
-            return next();
+    var populate = function(key, cb) {
+      if (debugEnabled) { debug('cache miss: ' + key); }
+      var cacheValue, contentType;
+
+      // replace write() to intercept the content sent to the client
+      resp._v_write = resp.write;
+      resp.write = function (chunk, encoding) {
+        resp._v_write(chunk, encoding);
+        if (chunk) {
+          if (cacheValue) {
+            cacheValue = undefined; // multiple writes, don't cache
+            resp.write = resp._v_write;
+          } else {
+            contentType = resp._headers['content-type'] || '';
+            cacheValue = chunk;
           }
         }
-      });
-    } else {
+      };
+
+      // replace end() to intercept the content returned to the client
+      var end = resp.end;
+      resp.end = function (chunk, encoding) {
+        resp.end = end;
+        if (chunk) {
+          if (cacheValue) {
+            cacheValue = undefined; // multiple writes, don't cache
+          } else {
+            resp.on('finish', function () {
+              contentType = resp._headers['content-type'] || '';
+              cacheValue = chunk;
+            });
+          }
+        }
+        resp.end(chunk, encoding);
+
+        cache(contentType, cacheValue, cb);
+      };
       return next();
-    }
+    };
+
+    resp.setHeader('Cache-Control', "public, max-age=" + Math.floor(options.ttl / 1000) + ", must-revalidate");
+    self.internalCache.getSet(key, populate, options, getSetCallback);
   };
 };
 
-function cache(self, key, options, contentType, chunk) {
-  chunk = chunk.toString();
-  var size = chunk.length + contentType.length + 1;
-  var buffer = new Buffer(size);
-  buffer.writeUInt8(contentType.length.valueOf(), 0);
-  buffer.write(contentType, 1);
-  buffer.write(chunk, contentType.length + 1);
-  self.internalCache.set(key, buffer, options, function (err) {
-    if (err) {
-      console.log('Cache error: ' + err);
-    } else if (debugEnabled) {
-      debug('Cached: ' + key);
-    }
-  });
+function cache(contentType, cacheValue, cb) {
+  var buffer;
+  if (cacheValue) {
+    cacheValue = cacheValue.toString();
+    var size = cacheValue.length + contentType.length + 1;
+    buffer = new Buffer(size);
+    buffer.writeUInt8(contentType.length.valueOf(), 0);
+    buffer.write(contentType, 1);
+    buffer.write(cacheValue, contentType.length + 1);
+  }
+  cb(null, buffer);
 }
@@ -123,0 +127,0 @@

lib/cache.js

@@ -27,2 +27,3 @@ /****************************************************************************
 var DEFAULT_TTL = 300;
+var eventEmitter = new (require('events').EventEmitter)();
 
@@ -42,2 +43,44 @@ function Cache(Spi, name, options) {
 
+// Retrieve an element from cache if present and set it using the provided populate function if not.
+// This method will also handle the "thundering herd" issue by coordinating waiting for already in-progress getSet()
+// requests and population for the same key.
+// The populate (key, callback) function must invoke its callback(error, reply) function on completion. If there is an
+// error, it must be passed as the first parameter (otherwise undefined or null). Assuming no error, the second
+// parameter is the value passed to the cache.
+// The options parameter contains any options to be passed as a part of the cache set() function.
+// The callback (error, reply, fromCache) function will be called after all processing has completed. It is called
+// immediately (fromCache == true) if the cache contains the item. Otherwise, it will be called once the
+// populate function has completed.
+// If "setEncoding" was previously called on this cache, then the value will be returned as a string
+// in the specified encoding. Otherwise, a Buffer will be returned.
+// key, populate, and callback are required. options is optional.
+Cache.prototype.getSet = function(key, populate, options, callback) {
+  validateKey(key);
+  if (!callback) { callback = options; options = {}; }
+  if (typeof populate !== 'function') { throw new Error('populate must be a function'); }
+  if (typeof callback !== 'function') { throw new Error('callback must be a function'); }
+
+  var self = this;
+  this.cache.get(key, function(err, reply) {
+    if (err || reply) {
+      if (reply && self.options.encoding) {
+        reply = reply.toString(self.options.encoding);
+      }
+      return callback(err, reply, true);
+    }
+
+    var event = self.name + key;
+    eventEmitter.once(event, callback);
+    if (eventEmitter.listeners(event).length > 1) { return; }
+
+    populate(key, function(err, reply) {
+      if (err) { return eventEmitter.emit(event, err, reply); }
+
+      self.set(key, reply, options, function(err) {
+        eventEmitter.emit(event, err, reply);
+      });
+    });
+  });
+};
+
 // Retrieve the element from cache and return as the second argument to "callback". (First argument is
@@ -44,0 +87,0 @@ // the error, or undefined if there is no error). It is an error to call this with no callback.

package.json

 {
   "name": "volos-cache-common",
-  "version": "0.9.4",
+  "version": "0.9.5",
   "main": "lib/cache.js",
@@ -5,0 +5,0 @@ "license": "MIT",

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

18480

increased by9.01%

Lines of code

369

increased by7.27%

No dependency changes