Skip to content

Commit 9481d8b

Browse files
alonlalonl
committed
Added cassandra backend
1 parent 3902eb0 commit 9481d8b

File tree

4 files changed

+218
-5
lines changed

4 files changed

+218
-5
lines changed

‎index.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,3 +2,4 @@ module.exports = require("./lib/acl.js");
22
module.exports.redisBackend = require("./lib/redis-backend.js");
33
module.exports.memoryBackend = require("./lib/memory-backend.js");
44
module.exports.mongodbBackend = require("./lib/mongodb-backend.js");
5+
module.exports.cassandraBackend = require("./lib/cassandra-backend.js");

‎lib/cassandra-backend.js‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,191 @@
1+
/**
2+
Cassandra Backend.
3+
Implementation of the storage backend using Cassandra
4+
5+
Attention: The specified keyspace and table must exist beforehand. The 'create table' query:
6+
CREATE TABLE [keyspace].[columnfamily] (
7+
bucketname varchar,
8+
key varchar,
9+
values set<varchar>,
10+
PRIMARY KEY ((bucketname, key))
11+
)
12+
*/
13+
"use strict";
14+
15+
var contract = require('./contract');
16+
var async = require('async');
17+
var _ = require('lodash');
18+
19+
function CassandraBackend(client, keyspace, columnfamily){
20+
this.client = client;
21+
var keyspaceWithTableName = keyspace + "." + columnfamily;
22+
this.queries = {
23+
clean: "TRUNCATE " + keyspaceWithTableName,
24+
get: "SELECT values FROM " + keyspaceWithTableName + " WHERE bucketname = ? AND key = ?",
25+
union: "SELECT values FROM " + keyspaceWithTableName + " WHERE bucketname = ? AND key IN ?",
26+
add: "UPDATE " + keyspaceWithTableName + " SET values = values + ? WHERE bucketname = ? AND key = ?",
27+
del: "DELETE FROM " + keyspaceWithTableName + " WHERE bucketname = ? AND key IN ?",
28+
remove: "UPDATE " + keyspaceWithTableName + " SET values = values - ? WHERE bucketname = ? AND key = ?"
29+
};
30+
}
31+
32+
CassandraBackend.prototype = {
33+
/**
34+
Begins a transaction.
35+
*/
36+
begin : function(){
37+
// returns a transaction object(just an array of functions will do here.)
38+
return [];
39+
},
40+
41+
/**
42+
Ends a transaction (and executes it)
43+
*/
44+
end : function(transaction, cb){
45+
contract(arguments).params('array', 'function').end();
46+
async.series(transaction,function(err){
47+
cb(err instanceof Error? err : undefined);
48+
});
49+
},
50+
51+
/**
52+
Cleans the whole storage.
53+
*/
54+
clean : function(cb){
55+
contract(arguments).params('function').end();
56+
this.client.execute(this.queries.clean, [], cb);
57+
},
58+
59+
/**
60+
Gets the contents at the bucket's key.
61+
*/
62+
get: function(bucket, key, cb) {
63+
contract(arguments)
64+
.params('string', 'string|number', 'function')
65+
.end();
66+
key = encodeText(key);
67+
this.client.execute(this.queries.get, [bucket, key], {hints: ['varchar', 'varchar']}, function(err, result) {
68+
if (err) return cb(err);
69+
if (result.rows.length == 0) return cb(undefined, []);
70+
result = decodeAll(result.rows[0].values);
71+
cb(undefined, result);
72+
});
73+
},
74+
75+
/**
76+
Returns the union of the values in the given keys.
77+
*/
78+
union: function(bucket, keys, cb) {
79+
contract(arguments)
80+
.params('string', 'array', 'function')
81+
.end();
82+
keys = encodeAll(keys);
83+
this.client.execute(this.queries.union, [bucket, keys], {hints: ['varchar', 'set<varchar>']}, function(err, result) {
84+
if (err) return cb(err);
85+
if (result.rows.length == 0) return cb(undefined, []);
86+
result = result.rows.reduce(function(prev, curr) { return prev.concat(decodeAll(curr.values)) }, []);
87+
cb(undefined, _.union(result));
88+
});
89+
},
90+
91+
/**
92+
Adds values to a given key inside a bucket.
93+
*/
94+
add: function(transaction, bucket, key, values) {
95+
contract(arguments)
96+
.params('array', 'string', 'string|number', 'string|array|number')
97+
.end();
98+
99+
if (key == "key") throw new Error("Key name 'key' is not allowed.");
100+
key = encodeText(key);
101+
var self = this;
102+
transaction.push(function (cb) {
103+
values = makeArray(values);
104+
self.client.execute(self.queries.add, [values, bucket, key], {hints: ['set<varchar>', 'varchar', 'varchar']}, function(err) {
105+
if (err) return cb(err);
106+
cb(undefined);
107+
});
108+
});
109+
},
110+
111+
/**
112+
Delete the given key(s) at the bucket
113+
*/
114+
del: function(transaction, bucket, keys) {
115+
contract(arguments)
116+
.params('array', 'string', 'string|array')
117+
.end();
118+
keys = makeArray(keys);
119+
var self = this;
120+
transaction.push(function (cb) {
121+
self.client.execute(self.queries.del, [bucket, keys], {hints: ['varchar', 'set<varchar>']}, function(err) {
122+
if (err) return cb(err);
123+
cb(undefined);
124+
});
125+
});
126+
},
127+
128+
/**
129+
Removes values from a given key inside a bucket.
130+
*/
131+
remove: function(transaction, bucket, key, values) {
132+
contract(arguments)
133+
.params('array', 'string', 'string|number', 'string|array|number')
134+
.end();
135+
key = encodeText(key);
136+
var self = this;
137+
values = makeArray(values);
138+
transaction.push(function (cb) {
139+
self.client.execute(self.queries.remove, [values, bucket, key], {hints: ['set<varchar>', 'varchar', 'varchar']}, function(err) {
140+
if (err) return cb(err);
141+
cb(undefined);
142+
});
143+
});
144+
}
145+
};
146+
147+
function encodeText(text) {
148+
if (typeof text == 'number' || text instanceof Number) text = text.toString();
149+
if (typeof text == 'string' || text instanceof String) {
150+
text = encodeURIComponent(text);
151+
text = text.replace(/\./, '%2E');
152+
}
153+
return text;
154+
}
155+
156+
function decodeText(text) {
157+
if (typeof text == 'string' || text instanceof String) {
158+
text = decodeURIComponent(text);
159+
}
160+
return text;
161+
}
162+
163+
function encodeAll(arr) {
164+
if (Array.isArray(arr)) {
165+
var ret = [];
166+
arr.forEach(function(aval) {
167+
ret.push(encodeText(aval));
168+
});
169+
return ret;
170+
} else {
171+
return arr;
172+
}
173+
}
174+
175+
function decodeAll(arr) {
176+
if (Array.isArray(arr)) {
177+
var ret = [];
178+
arr.forEach(function(aval) {
179+
ret.push(decodeText(aval));
180+
});
181+
return ret;
182+
} else {
183+
return arr;
184+
}
185+
}
186+
187+
function makeArray(arr){
188+
return Array.isArray(arr) ? encodeAll(arr) : [encodeText(arr)];
189+
}
190+
191+
exports = module.exports = CassandraBackend;

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
"dependencies": {
1818
"async": "~0.9.0",
1919
"bluebird": "^2.3.11",
20+
"cassandra-driver": "^2.0.1",
2021
"lodash": "~2.4.1",
2122
"mongodb": "^1.4.30",
2223
"redis": ">=0.12.1"

‎test/runner.js‎

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,26 @@ describe('MongoDB - useSingle', function () {
3434
run()
3535
});
3636

37+
// Attention: keyspace and columnfamily must exist beforehand
38+
describe('Cassandra', function () {
39+
before(function (done) {
40+
var self = this
41+
, cassandra = require('cassandra-driver');
42+
43+
client = new cassandra.Client({contactPoints: ['127.0.0.1']});
44+
client.connect(function(err) {
45+
if (err) return done(err);
46+
client.execute("TRUNCATE acltest.acl", [], function(err) {
47+
if (err) return done(err);
48+
self.backend = new Acl.cassandraBackend(client, "acltest", "acl");
49+
done()
50+
});
51+
});
52+
});
53+
54+
run()
55+
});
56+
3757
describe('Redis', function () {
3858
before(function (done) {
3959
var self = this
@@ -43,22 +63,22 @@ describe('Redis', function () {
4363
password: null
4464
}
4565
, Redis = require('redis')
46-
47-
66+
67+
4868
var redis = Redis.createClient(options.port, options.host, {no_ready_check: true} )
4969

5070
function start(){
5171
self.backend = new Acl.redisBackend(redis)
5272
done()
5373
}
54-
74+
5575
if (options.password) {
5676
redis.auth(options.password, start)
5777
} else {
5878
start()
5979
}
6080
})
61-
81+
6282
run()
6383
})
6484

@@ -68,7 +88,7 @@ describe('Memory', function () {
6888
var self = this
6989
self.backend = new Acl.memoryBackend()
7090
})
71-
91+
7292
run()
7393
})
7494

0 commit comments

Comments
 (0)