Merge pull request #1031 from strongloop/kv-memory-2x

Implement KeyValue API and memory connector [2.x]

index.js

@@ -19,3 +19,5 @@ Object.defineProperty(exports, 'test', {
 });
 
 exports.Transaction = require('loopback-connector').Transaction;
+
+exports.KeyValueAccessObject = require('./lib/kvao');

lib/connectors/kv-memory.js

@@ -0,0 +1,143 @@
+'use strict';
+
+var assert = require('assert');
+var Connector = require('loopback-connector').Connector;
+var debug = require('debug')('loopback:connector:kv-memory');
+var util = require('util');
+
+exports.initialize = function initializeDataSource(dataSource, cb) {
+  var settings = dataSource.settings;
+  dataSource.connector = new KeyValueMemoryConnector(settings, dataSource);
+  if (cb) process.nextTick(cb);
+};
+
+function KeyValueMemoryConnector(settings, dataSource) {
+  Connector.call(this, 'kv-memory', settings);
+
+  debug('Connector settings', settings);
+
+  this.dataSource = dataSource;
+  this.DataAccessObject = dataSource.juggler.KeyValueAccessObject;
+
+  this._store = Object.create(null);
+
+  this._setupRegularCleanup();
+};
+util.inherits(KeyValueMemoryConnector, Connector);
+
+KeyValueMemoryConnector.prototype._setupRegularCleanup = function() {
+  // Scan the database for expired keys at a regular interval
+  // in order to release memory. Note that GET operation checks
+  // key expiration too, the scheduled cleanup is merely a performance
+  // optimization.
+  var self = this;
+  this._cleanupTimer = setInterval(
+    function() { self._removeExpiredItems(); },
+    1000);
+  this._cleanupTimer.unref();
+};
+
+KeyValueMemoryConnector._removeExpiredItems = function() {
+  debug('Running scheduled cleanup of expired items.');
+  for (var modelName in this._store) {
+    var modelStore = this._store[modelName];
+    for (var key in modelStore) {
+      if (modelStore[key].isExpired()) {
+        debug('Removing expired key', key);
+        delete modelStore[key];
+      }
+    }
+  }
+};
+
+KeyValueMemoryConnector.prototype._getStoreForModel = function(modelName) {
+  if (!(modelName in this._store)) {
+    this._store[modelName] = Object.create(null);
+  }
+  return this._store[modelName];
+};
+
+KeyValueMemoryConnector.prototype.get =
+function(modelName, key, options, callback) {
+  var store = this._getStoreForModel(modelName);
+  var item = store[key];
+
+  if (item && item.isExpired()) {
+    debug('Removing expired key', key);
+    delete store[key];
+    item = undefined;
+  }
+
+  var value = item ? item.value : null;
+
+  debug('GET %j %j -> %s', modelName, key, value);
+
+  if (/^buffer:/.test(value)) {
+    value = new Buffer(value.slice(7), 'base64');
+  } else if (/^date:/.test(value)) {
+    value = new Date(value.slice(5));
+  } else if (value != null) {
+    value = JSON.parse(value);
+  }
+
+  process.nextTick(function() {
+    callback(null, value);
+  });
+};
+
+KeyValueMemoryConnector.prototype.set =
+function(modelName, key, value, options, callback) {
+  var store = this._getStoreForModel(modelName);
+  var value;
+  if (Buffer.isBuffer(value)) {
+    value = 'buffer:' + value.toString('base64');
+  } else if (value instanceof Date) {
+    value = 'date:' + value.toISOString();
+  } else {
+    value = JSON.stringify(value);
+  }
+
+  debug('SET %j %j %s %j', modelName, key, value, options);
+  store[key] = new StoreItem(value, options && options.ttl);
+
+  process.nextTick(callback);
+};
+
+KeyValueMemoryConnector.prototype.expire =
+function(modelName, key, ttl, options, callback) {
+  var store = this._getStoreForModel(modelName);
+
+  if (!(key in store)) {
+    return process.nextTick(function() {
+      callback(new Error('Cannot expire unknown key ' + key));
+    });
+  }
+
+  debug('EXPIRE %j %j %s', modelName, key, ttl || '(never)');
+  store[key].setTtl(ttl);
+  process.nextTick(callback);
+};
+
+KeyValueMemoryConnector.prototype.disconnect = function(callback) {
+  if (this._cleanupTimer)
+    clearInterval(this._cleanupTimer);
+  this._cleanupTimer = null;
+  process.nextTick(callback);
+};
+
+function StoreItem(value, ttl) {
+  this.value = value;
+  this.setTtl(ttl);
+}
+
+StoreItem.prototype.isExpired = function() {
+  return this.expires && this.expires <= Date.now();
+};
+
+StoreItem.prototype.setTtl = function(ttl) {
+  if (ttl) {
+    this.expires = Date.now() + ttl;
+  } else {
+    this.expires = undefined;
+  }
+};

lib/datasource.js

@@ -20,6 +20,7 @@ var util = require('util');
 var assert = require('assert');
 var async = require('async');
 var traverse = require('traverse');
+var juggler = require('..');
 
 if (process.env.DEBUG === 'loopback') {
   // For back-compatibility
@@ -106,6 +107,7 @@ function DataSource(name, settings, modelBuilder) {
   this.modelBuilder = modelBuilder || new ModelBuilder();
   this.models = this.modelBuilder.models;
   this.definitions = this.modelBuilder.definitions;
+  this.juggler = juggler;
 
   // operation metadata
   // Initialize it before calling setup as the connector might register operations

lib/kvao/expire.js

@@ -0,0 +1,33 @@
+'use strict';
+
+var assert = require('assert');
+var utils = require('../utils');
+
+/**
+ * Set expiration (TTL) for the given key.
+ *
+ * @param {String} key
+ * @param {Number} ttl
+ * @param {Object} options
+ * @callback cb
+ * @param {Error} error
+ *
+ * @header KVAO.get(key, cb)
+ */
+module.exports = function keyValueExpire(key, ttl, options, callback) {
+  if (callback == undefined && typeof options === 'function') {
+    callback = options;
+    options = {};
+  } else if (!options) {
+    options = {};
+  }
+
+  assert(typeof key === 'string' && key, 'key must be a non-empty string');
+  assert(typeof ttl === 'number' && ttl > 0, 'ttl must be a positive integer');
+  assert(typeof options === 'object', 'options must be an object');
+
+  callback = callback || utils.createPromiseCallback();
+  this.getConnector().expire(this.modelName, key, ttl, options, callback);
+  return callback.promise;
+};
+

lib/kvao/get.js

@@ -0,0 +1,32 @@
+'use strict';
+
+var assert = require('assert');
+var utils = require('../utils');
+
+/**
+ * Get the value stored for the given key.
+ *
+ * @param {String} key
+ * @callback cb
+ * @param {Error} error
+ * @param {*} value
+ *
+ * @header KVAO.get(key, cb)
+ */
+module.exports = function keyValueGet(key, options, callback) {
+  if (callback == undefined && typeof options === 'function') {
+    callback = options;
+    options = {};
+  } else if (!options) {
+    options = {};
+  }
+
+  assert(typeof key === 'string' && key, 'key must be a non-empty string');
+
+  callback = callback || utils.createPromiseCallback();
+  this.getConnector().get(this.modelName, key, options, function(err, result) {
+    // TODO convert raw result to Model instance (?)
+    callback(err, result);
+  });
+  return callback.promise;
+};

lib/kvao/index.js

@@ -0,0 +1,15 @@
+'use strict';
+
+function KeyValueAccessObject() {
+};
+
+module.exports = KeyValueAccessObject;
+
+KeyValueAccessObject.get = require('./get');
+KeyValueAccessObject.set = require('./set');
+KeyValueAccessObject.expire = require('./expire');
+
+KeyValueAccessObject.getConnector = function() {
+  return this.getDataSource().connector;
+};
+

lib/kvao/set.js

@@ -0,0 +1,40 @@
+'use strict';
+
+var assert = require('assert');
+var utils = require('../utils');
+
+/**
+ * Set the value for the given key.
+ *
+ * @param {String} key
+ * @param {*} value
+ * @callback cb
+ * @param {Error} error
+ *
+ * @header KVAO.set(key, value, cb)
+ */
+module.exports = function keyValueSet(key, value, options, callback) {
+  if (callback == undefined && typeof options === 'function') {
+    callback = options;
+    options = {};
+  } else if (typeof options === 'number') {
+    options = { ttl: options };
+  } else if (!options) {
+    options = {};
+  }
+
+  assert(typeof key === 'string' && key, 'key must be a non-empty string');
+  assert(value != null, 'value must be defined and not null');
+  assert(typeof options === 'object', 'options must be an object');
+  if (options && 'ttl' in options) {
+    assert(typeof options.ttl === 'number' && options.ttl > 0,
+      'options.ttl must be a positive number');
+  }
+
+  callback = callback || utils.createPromiseCallback();
+
+  // TODO convert possible model instance in "value" to raw data via toObect()
+  this.getConnector().set(this.modelName, key, value, options, callback);
+  return callback.promise;
+};
+

test/kv-memory.js

@@ -0,0 +1,16 @@
+var kvMemory = require('../lib/connectors/kv-memory');
+var DataSource = require('..').DataSource;
+
+describe('KeyValue-Memory connector', function() {
+  var lastDataSource;
+  var dataSourceFactory = function() {
+    lastDataSource = new DataSource({ connector: kvMemory });
+    return lastDataSource;
+  };
+
+  afterEach(function disconnectKVMemoryConnector() {
+    if (lastDataSource) return lastDataSource.disconnect();
+  });
+
+  require('./kvao.suite')(dataSourceFactory);
+});

test/kvao.suite.js

@@ -0,0 +1,23 @@
+'use strict';
+
+var debug = require('debug')('test');
+var fs = require('fs');
+var path = require('path');
+
+module.exports = function(dataSourceFactory, connectorCapabilities) {
+  describe('KeyValue API', function loadAllTestFiles() {
+    var testRoot = path.resolve(__dirname, 'kvao');
+    var testFiles = fs.readdirSync(testRoot);
+    testFiles = testFiles.filter(function(it) {
+      return !!require.extensions[path.extname(it).toLowerCase()] &&
+        /\.suite\.[^.]+$/.test(it);
+    });
+
+    for (var ix in testFiles) {
+      var name = testFiles[ix];
+      var fullPath = path.resolve(testRoot, name);
+      debug('Loading test suite %s (%s)', name, fullPath);
+      require(fullPath)(dataSourceFactory, connectorCapabilities);
+    }
+  });
+};

test/kvao/_helpers.js

@@ -0,0 +1,9 @@
+'use strict';
+
+exports.givenCacheItem = function(dataSourceFactory) {
+  var dataSource = dataSourceFactory();
+  return dataSource.createModel('CacheItem', {
+    key: String,
+    value: 'any',
+  });
+};

test/kvao/expire.suite.js

@@ -0,0 +1,44 @@
+'use strict';
+
+var should = require('should');
+var helpers = require('./_helpers');
+var Promise = require('bluebird');
+
+module.exports = function(dataSourceFactory, connectorCapabilities) {
+  describe('expire', function() {
+    var CacheItem;
+    beforeEach(function unpackContext() {
+      CacheItem = helpers.givenCacheItem(dataSourceFactory);
+    });
+
+    it('sets key ttl - Callback API', function(done) {
+      CacheItem.set('a-key', 'a-value', function(err) {
+        if (err) return done(err);
+        CacheItem.expire('a-key', 1, function(err) {
+          if (err) return done(err);
+          setTimeout(function() {
+            CacheItem.get('a-key', function(err, value) {
+              if (err) return done(err);
+              should.equal(value, null);
+              done();
+            });
+          }, 20);
+        });
+      });
+    });
+
+    it('sets key ttl - Promise API', function() {
+      return Promise.resolve(CacheItem.set('a-key', 'a-value'))
+        .then(function() { return CacheItem.expire('a-key', 1); })
+        .delay(20)
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { should.equal(value, null); });
+    });
+
+    it('returns error when key does not exist', function() {
+      return CacheItem.expire('key-does-not-exist', 1).then(
+        function() { throw new Error('expire() should have failed'); },
+        function(err) { err.message.should.match(/key-does-not-exist/); });
+    });
+  });
+};

test/kvao/get-test.suite.js

@@ -0,0 +1,103 @@
+'use strict';
+
+var should = require('should');
+var helpers = require('./_helpers');
+var Promise = require('bluebird');
+
+module.exports = function(dataSourceFactory, connectorCapabilities) {
+  describe('get/set', function() {
+    var CacheItem;
+    beforeEach(function unpackContext() {
+      CacheItem = helpers.givenCacheItem(dataSourceFactory);
+    });
+
+    it('works for string values - Callback API', function(done) {
+      CacheItem.set('a-key', 'a-value', function(err) {
+        if (err) return done(err);
+        CacheItem.get('a-key', function(err, value) {
+          if (err) return done(err);
+          should.equal(value, 'a-value');
+          done();
+        });
+      });
+    });
+
+    it('works for string values - Promise API', function() {
+      return CacheItem.set('a-key', 'a-value')
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { should.equal(value, 'a-value'); });
+    });
+
+    it('works for Object values', function() {
+      return CacheItem.set('a-key', { a: 1, b: 2 })
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { value.should.eql({ a: 1, b: 2 }); });
+    });
+
+    it('works for Buffer values', function() {
+      return CacheItem.set('a-key', new Buffer([1, 2, 3]))
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { value.should.eql(new Buffer([1, 2, 3])); });
+    });
+
+    it('works for Date values', function() {
+      return CacheItem.set('a-key', new Date('2016-08-03T11:53:03.470Z'))
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) {
+          value.should.be.instanceOf(Date);
+          value.toISOString().should.equal('2016-08-03T11:53:03.470Z');
+        });
+    });
+
+    it('works for Number values - integers', function() {
+      return CacheItem.set('a-key', 12345)
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { value.should.equal(12345); });
+    });
+
+    it('works for Number values - floats', function() {
+      return CacheItem.set('a-key', 12.345)
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { value.should.equal(12.345); });
+    });
+
+    it('works for Boolean values', function() {
+      return CacheItem.set('a-key', false)
+        .then(function() { return CacheItem.get('a-key'); })
+        .then(function(value) { value.should.equal(false); });
+    });
+
+    it('honours options.ttl', function() {
+      return Promise.resolve(CacheItem.set('a-key', 'a-value', { ttl: 10 }))
+      .delay(20)
+      .then(function() { return CacheItem.get('a-key'); })
+      .then(function(value) { should.equal(value, null); });
+    });
+
+    describe('get', function() {
+      it('returns "null" when key does not exist', function() {
+        return CacheItem.get('key-does-not-exist')
+          .then(function(value) { should.equal(value, null); });
+      });
+
+      it('converts numeric options arg to options.ttl', function() {
+        return Promise.resolve(CacheItem.set('a-key', 'a-value', 10))
+          .delay(20)
+          .then(function() { return CacheItem.get('a-key'); })
+          .then(function(value) { should.equal(value, null); });
+      });
+    });
+
+    describe('set', function() {
+      it('resets TTL timer', function() {
+        return Promise.resolve(CacheItem.set('a-key', 'a-value', { ttl: 10 }))
+          .then(function() {
+            return CacheItem.set('a-key', 'another-value'); // no TTL
+          })
+          .delay(20)
+          .then(function() { return CacheItem.get('a-key'); })
+          .then(function(value) { should.equal(value, 'another-value'); });
+      });
+    });
+  });
+};