Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

initial work for compare and swap #67

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

initial work for compare and swap #67

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 21 additions & 15 deletions lib/memjs/memjs.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ Client.prototype.server = function(key) {
//
// Fetches the value at the given key with callback signature:
//
// callback(err, value, flags)
// callback(err, value, flags, cas)
//
// _value_ and _flags_ are both `Buffer`s.
// If the key is not found, the callback is invoked
Expand All @@ -110,20 +110,20 @@ Client.prototype.get = function(key, callback) {
var logger = this.options.logger;
this.perform(key, request, function(err, response) {
if (err) {
callback && callback(err, null, null);
callback && callback(err, null, null, null);
return;
}
switch (response.header.status) {
case 0:
callback && callback(null, response.val, response.extras)
callback && callback(null, response.val, response.extras, response.header.cas)
break;
case 1:
callback && callback(null, null, null);
callback && callback(null, null, null, null);
break;
default:
var errorMessage = 'MemJS GET: ' + errors[response.header.status];
logger.log(errorMessage);
callback && callback(new Error(errorMessage), null, null);
callback && callback(new Error(errorMessage), null, null, null);
}
});
}
Expand All @@ -134,12 +134,12 @@ Client.prototype.get = function(key, callback) {
// in last, after the callback) overrides the default expiration (see
// `Client.create`). The callback signature is:
//
// callback(err, success)
Client.prototype.set = function(key, value, callback, expires) {
// callback(err, success, cas)
Client.prototype.set = function(key, value, callback, expires, cas) {
var extras = Buffer.concat([new Buffer('00000000', 'hex'),
makeExpiration(expires || this.options.expires)]);
this.seq++;
var request = makeRequestBuffer(1, key, extras, value, this.seq);
var request = makeRequestBuffer(1, key, extras, value, this.seq, cas);
var logger = this.options.logger;
this.perform(key, request, function(err, response) {
if (err) {
Expand All @@ -148,7 +148,10 @@ Client.prototype.set = function(key, value, callback, expires) {
}
switch (response.header.status) {
case 0:
callback && callback(null, true)
callback && callback(null, true, response.header.cas);
break;
case 2: // Key exists (CAS didn't match)
callback && callback(null, false, null);
break;
default:
var errorMessage = 'MemJS SET: ' + errors[response.header.status];
Expand Down Expand Up @@ -178,10 +181,10 @@ Client.prototype.add = function(key, value, callback, expires) {
}
switch (response.header.status) {
case 0:
callback && callback(null, true)
callback && callback(null, true, response.header.cas)
break;
case 2:
callback && callback(null, false);
callback && callback(null, false, null);
break;
default:
var errorMessage = 'MemJS ADD: ' + errors[response.header.status];
Expand All @@ -199,10 +202,10 @@ Client.prototype.add = function(key, value, callback, expires) {
// `Client.create`). The callback signature is:
//
// callback(err, success)
Client.prototype.replace = function(key, value, callback, expires) {
Client.prototype.replace = function(key, value, callback, expires, cas) {
var extras = Buffer.concat([new Buffer('00000000', 'hex'), makeExpiration(expires || this.options.expires)]);
this.seq++;
var request = makeRequestBuffer(3, key, extras, value, this.seq);
var request = makeRequestBuffer(3, key, extras, value, this.seq, cas);
var logger = this.options.logger;
this.perform(key, request, function(err, response) {
if (err) {
Expand All @@ -211,10 +214,13 @@ Client.prototype.replace = function(key, value, callback, expires) {
}
switch (response.header.status) {
case 0:
callback && callback(null, true)
callback && callback(null, true, response.header.cas);
break;
case 1:
callback && callback(null, false);
callback && callback(null, false, null);
break;
case 2:
callback && callback(null, false, response.header.cas);
break;
default:
var errorMessage = 'MemJS REPLACE: ' + errors[response.header.status];
Expand Down
5 changes: 3 additions & 2 deletions lib/memjs/utils.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ var bufferify = function(val) {
return Buffer.isBuffer(val) ? val : new Buffer(val);
}

exports.makeRequestBuffer = function(opcode, key, extras, value, opaque) {
exports.makeRequestBuffer = function(opcode, key, extras, value, opaque, cas) {
key = bufferify(key);
extras = bufferify(extras);
value = bufferify(value);
Expand All @@ -16,7 +16,8 @@ exports.makeRequestBuffer = function(opcode, key, extras, value, opaque) {
keyLength: key.length,
extrasLength: extras.length,
totalBodyLength: key.length + value.length + extras.length,
opaque: opaque
opaque: opaque,
cas: cas
};
header.toBuffer(requestHeader).copy(buf);
extras.copy(buf, 24)
Expand Down
41 changes: 38 additions & 3 deletions test/client_test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,21 @@ exports.testGetSuccessful = function(beforeExit, assert) {
assert.equal('hello', request.key);
n += 1;
dummyServer.respond(
{header: {status: 0, opaque: request.header.opaque},
val: 'world', extras: 'flagshere'});
{
header: {
status: 0,
opaque: request.header.opaque,
cas: new Buffer([0x0a, 0, 0, 0, 0, 0, 0, 0])
},
val: 'world', extras: 'flagshere'
});
}

var client = new MemJS.Client([dummyServer]);
client.get('hello', function(err, val, flags) {
client.get('hello', function(err, val, flags, cas) {
assert.equal('world', val);
assert.equal('flagshere', flags);
assert.deepEqual(new Buffer([0x0a, 0, 0, 0, 0, 0, 0, 0]), cas);
assert.equal(null, err);
callbn += 1;
});
Expand Down Expand Up @@ -104,6 +111,34 @@ exports.testSetWithExpiration = function(beforeExit, assert) {
});
}

exports.testSetWithCas = function(beforeExit, assert) {
var n = 0;
var callbn = 0;
var dummyServer = new MemJS.Server();
dummyServer.write = function(requestBuf) {
console.log(requestBuf);
request = MemJS.Utils.parseMessage(requestBuf);
assert.equal('hello', request.key);
assert.equal('world', request.val);
assert.deepEqual(new Buffer([0x0a,0,0,0,0,0,0,0]), request.header.cas);
n += 1;
dummyServer.respond({header: {status: 0, opaque: request.header.opaque}});
}

var client = new MemJS.Client([dummyServer]);
client.set('hello', 'world', function(err, val) {
assert.equal(null, err);
assert.equal(true, val);
callbn += 1;
}, undefined, new Buffer([0x0a, 0, 0, 0, 0, 0, 0, 0]));


beforeExit(function() {
assert.equal(1, n, 'Ensure set is called');
assert.equal(1, callbn, 'Ensure callback is called');
});
}

exports.testSetUnsuccessful = function(beforeExit, assert) {
var n = 0;
var callbn = 0;
Expand Down