Merge branch 'parallel-requests-fix' into develop

lib/caching.js

@@ -18,6 +18,8 @@ var caching = function (args) {
     // do we handle a cache error the same as a cache miss?
     self.ignoreCacheErrors = args.ignoreCacheErrors || false;
 
+    self.queues = {};
+
     /**
      * Wraps a function in cache. I.e., the first time the function is run,
      * its results are stored in cache so subsequent calls retrieve from cache
@@ -34,21 +36,33 @@ var caching = function (args) {
      */
     self.wrap = function (key, work, cb) {
         self.store.get(key, function (err, result) {
-            if (err && (!self.ignoreCacheErrors)) { return cb(err); }
-            if (result) {
-                return cb(null, result);
-            }
+            if (err && (!self.ignoreCacheErrors)) {
+                cb(err);
+            } else if (result) {
+                cb.apply(null, result);
+            } else if (self.queues[key]) {
+                self.queues[key].push(cb);
+            } else {
+                self.queues[key] = [cb];
+
+                work(function () {
+                    var work_args = Array.prototype.slice.call(arguments, 0);
+                    if (work_args[0]) { // assume first arg is an error
+                        return cb(work_args[0]);
+                    }
+                    self.store.set(key, work_args, function (err) {
+                        if (err && (!self.ignoreCacheErrors)) {
+                            return cb(err);
+                        }
 
-            work(function () {
-                var work_args = Array.prototype.slice.call(arguments, 0);
-                if (work_args[0]) { // assume first arg is an error
-                    return cb(work_args[0]);
-                }
-                self.store.set(key, work_args[1], function (err) {
-                    if (err && (!self.ignoreCacheErrors)) { return cb(err); }
-                    cb.apply(null, work_args);
+                        self.queues[key].forEach(function (done) {
+                            done.apply(null, work_args);
+                        });
+
+                        delete self.queues[key];
+                    });
                 });
-            });
+            }
         });
     };
 

test/caching.unit.js

@@ -235,35 +235,49 @@ describe("caching", function () {
                 });
             });
 
-            it("retrieves data from memory when available", function (done) {
-                cache.wrap(key, function (cb) {
-                    methods.get_widget(name, cb);
-                }, function (err, widget) {
-                    check_err(err);
-                    assert.ok(widget);
+            context("when result is already cached", function () {
+                function get_cached_widget(name, cb) {
+                    cache.wrap(key, function (cache_cb) {
+                        methods.get_widget(name, cache_cb);
+                    }, cb);
+                }
 
-                    memory_store_stub.get(key, function (err, result) {
+                beforeEach(function (done) {
+                    get_cached_widget(name, function (err, widget) {
                         check_err(err);
-                        assert.ok(result);
+                        assert.ok(widget);
 
-                        sinon.spy(memory_store_stub, 'get');
-                        var func_called = false;
-
-                        cache.wrap(key, function (cb) {
-                            methods.get_widget(name, function (err, result) {
-                                func_called = true;
-                                cb(err, result);
-                            });
-                        }, function (err, widget) {
+                        memory_store_stub.get(key, function (err, result) {
                             check_err(err);
-                            assert.deepEqual(widget, {name: name});
-                            assert.ok(memory_store_stub.get.calledWith(key));
-                            assert.ok(!func_called);
-                            memory_store_stub.get.restore();
+                            assert.ok(result);
+
+                            sinon.spy(memory_store_stub, 'get');
+
                             done();
                         });
                     });
                 });
+
+                afterEach(function () {
+                    memory_store_stub.get.restore();
+                });
+
+                it("retrieves data from cache", function (done) {
+                    var func_called = false;
+
+                    cache.wrap(key, function (cb) {
+                        methods.get_widget(name, function (err, result) {
+                            func_called = true;
+                            cb(err, result);
+                        });
+                    }, function (err, widget) {
+                        check_err(err);
+                        assert.deepEqual(widget, {name: name});
+                        assert.ok(memory_store_stub.get.calledWith(key));
+                        assert.ok(!func_called);
+                        done();
+                    });
+                });
             });
 
             it("expires cached result after ttl seconds", function (done) {
@@ -393,6 +407,45 @@ describe("caching", function () {
                 });
             });
         });
+
+        describe("when called multiple times in parallel with same key", function () {
+            var construct;
+
+            beforeEach(function () {
+                cache = caching({
+                    store: 'memory',
+                    max: 50,
+                    ttl: 5 * 60
+                });
+
+                construct = sinon.spy(function (val, cb) {
+                    var timeout = support.random.number(100);
+                    setTimeout(function () {
+                        cb(null, 'value');
+                    }, timeout);
+                });
+            });
+
+            it("calls the wrapped function once", function (done) {
+                var values = [];
+                for (var i = 0; i < 2; i++) {
+                    values.push(i);
+                }
+
+                async.each(values, function (val, async_cb) {
+                    cache.wrap('key', function (cb) {
+                        construct(val, cb);
+                    }, function (err, result) {
+                        assert.equal(result, 'value');
+                        async_cb(err);
+                    });
+                }, function (err) {
+                    check_err(err);
+                    assert.equal(construct.callCount, 1);
+                    done();
+                });
+            });
+        });
     });
 
     describe("instantiating with no store passed in", function () {

test/support.js

@@ -13,6 +13,11 @@ var support = {
                 random_str += chars.substring(rnum, rnum + 1);
             }
             return random_str;
+        },
+
+        number: function (max) {
+            max = max || 1000;
+            return Math.floor((Math.random() * max));
         }
     },