Repository: basho/riak-nodejs-client
Pull request #28 (2015-03-28 14:52:25)

CLIENTS-291: Changed BaseCommand and command signatures.
The BaseCommand class now takes the callback as a third arg and
verifies it is a function. It then provides this to subclasses
via _callback()

All commands are changed to have their constructors with the
signature: command(options, callback) and pass the callback to
the BaseCommand constructor.


Patch:
diff --git a/lib/commands/commandbase.js b/lib/commands/commandbase.js
index d782467..574880a 100644
--- a/lib/commands/commandbase.js
+++ b/lib/commands/commandbase.js
@@ -39,7 +39,7 @@ var logger = require('winston');
  * @param {String} pbRequestName name of the Riak protocol buffer this command will send
  * @param {String} pbResponseName name of the Riak protocol buffer this command will receive
  */
-function CommandBase(pbRequestName, pbResponseName) {
+function CommandBase(pbRequestName, pbResponseName, callback) {
     
     var requestCode = ProtoBufFactory.getCodeFor(pbRequestName);
     this.expectedCode = ProtoBufFactory.getCodeFor(pbResponseName);
@@ -49,6 +49,18 @@ function CommandBase(pbRequestName, pbResponseName) {
         logger.debug("No pbBuilder for " + pbRequestName);
     }
    
+    var schema = Joi.func().required();
+    var self = this;
+    Joi.validate(callback, schema, function(err, option) {
+       
+       if (err) {
+           throw 'callback is required and must be a function.';
+       }
+       
+       self.callback = callback;
+        
+    });
+   
     this.header = new Buffer(5);
     this.header.writeUInt8(requestCode, 4);
     
@@ -60,6 +72,15 @@ function CommandBase(pbRequestName, pbResponseName) {
 
 }
 
+/**
+ * Fires the user's callback with the arguments passed in.
+ * @method getCallback
+ * @returns {Function} the user supplied callback
+ */
+CommandBase.prototype._callback = function() {
+    this.callback.apply(this, arguments);
+};
+
 /**
  * Returns the expected response code for this command.
  * @method getExpectedResponseCode
@@ -109,7 +130,7 @@ CommandBase.prototype.getPbReqBuilder = function() {
 
 /** 
  * Construct and return the Riak protocol buffer message for this command.
- * Commands must override this method.
+ * Subclasses must override this method.
  * @method constructPbRequest
  * @returns {Object} a protocol buffer message builder
  */
@@ -119,7 +140,7 @@ CommandBase.prototype.constructPbRequest = function() {
 
 /**
  * Called by RiakNode when a response is received.
- * Commands must override this method.
+ * Subclasses must override this method.
  * @method onSuccess
  * @param {Object} pbResponseMessage the protocol buffer received from riak
  * @returns {Boolean} true if not streaming or the last response has been received, false otherwise.
@@ -135,7 +156,7 @@ CommandBase.prototype.onSuccess = function(pbResponseMessage) {
  * @param {Object} rpbErrorResp the RpbErrorResp protocol buffer
  */
 CommandBase.prototype.onRiakError = function(rpbErrorResp) {
-    throw 'Not supported yet!';
+    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
 };
 
 /**
@@ -144,7 +165,7 @@ CommandBase.prototype.onRiakError = function(rpbErrorResp) {
  * @param {String} msg an error message
  */
 CommandBase.prototype.onError = function(msg) {
-    throw 'Not supported yet!';
+    this._callback(msg, null);
 };
 
 // TODO how to inherit from this schema in derived classes?
diff --git a/lib/commands/crdt/fetchcounter.js b/lib/commands/crdt/fetchcounter.js
index 67f8b41..0acf59b 100644
--- a/lib/commands/crdt/fetchcounter.js
+++ b/lib/commands/crdt/fetchcounter.js
@@ -43,18 +43,19 @@ var Joi = require('joi');
  * @param {String} options.bucketType the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
  * @param {String} options.key the key for the counter you want to fetch.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Number} options.callback.response the counter value from Riak.
  * @param {Number} [options.timeout] set a timeout for this operation.
  * @param {Number} [options.r] the R value to use for this fetch.
  * @param {Number} [options.pr] the PR value to use for this fetch.
  * @param {Boolean} [options.notFoundOk] if true a vnode returning notfound for a key increments the r tally.
  * @param {Boolean} [options.useBasicQuorum] controls whether a read request should return early in some fail cases.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Number} callback.response the counter value from Riak.
+
  * @extends CommandBase 
  */
-function FetchCounter(options) {
-    CommandBase.call(this, 'DtFetchReq', 'DtFetchResp');
+function FetchCounter(options, callback) {
+    CommandBase.call(this, 'DtFetchReq', 'DtFetchResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -94,23 +95,14 @@ FetchCounter.prototype.onSuccess = function(dtFetchResp) {
         // sint64 is weird. You either get the zigzag encoding, or
         // you can get the actual number via the bytebuffer.
         var value = dtFetchResp.value.counter_value.toNumber();
-        this.options.callback(null, value);
+        this._callback(null, value);
     } else {
-        this.options.callback(null, null);
+        this._callback(null, null);
     }
     
     return true;
 };
 
-FetchCounter.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-FetchCounter.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
    bucket: Joi.string().required(),
    bucketType: Joi.string().required(),
@@ -120,7 +112,7 @@ var schema = Joi.object().keys({
    notFoundOk: Joi.boolean().default(null).optional(),
    basicQuorum: Joi.boolean().default(null).optional(),
    timeout: Joi.number().default(null).optional(),
-   callback: Joi.func().required()
+   callback: Joi.func().strip().optional()
 });
 
 
@@ -251,7 +243,7 @@ Builder.prototype = {
      * @return {FetchCounter}
      */
     build : function() {
-        return new FetchCounter(this);
+        return new FetchCounter(this, this.callback);
     }
     
 };
diff --git a/lib/commands/crdt/fetchset.js b/lib/commands/crdt/fetchset.js
index b074044..998b8ff 100644
--- a/lib/commands/crdt/fetchset.js
+++ b/lib/commands/crdt/fetchset.js
@@ -42,19 +42,20 @@ var Joi = require('joi');
  * @param {String} options.bucketType the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
  * @param {String} options.key the key for the counter you want to fetch.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Object} options.callback.response the set from Riak. null if not found.
  * @param {Number} [options.timeout] set a timeout for this operation.
  * @param {Number} [options.r] the R value to use for this fetch.
  * @param {Number} [options.pr] the PR value to use for this fetch.
  * @param {Boolean} [options.notFoundOk] if true a vnode returning notfound for a key increments the r tally.
  * @param {Boolean} [options.useBasicQuorum] controls whether a read request should return early in some fail cases.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Object} callback.response the set from Riak. null if not found.
+
  * @extends CommandBase 
  */
 
-function FetchSet(options) {
-    CommandBase.call(this, 'DtFetchReq', 'DtFetchResp');
+function FetchSet(options, callback) {
+    CommandBase.call(this, 'DtFetchReq', 'DtFetchResp', callback);
 
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -110,20 +111,12 @@ FetchSet.prototype.onSuccess = function(dtFetchResp) {
             value: valueStrings
         };
     }
-    this.options.callback(null, response);
+    this._callback(null, response);
     return true;
 };
 
-FetchSet.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-
-FetchSet.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
 
     // namespace
     bucket: Joi.string().required(),
@@ -169,7 +162,7 @@ Builder.prototype = {
      * @return {FetchSet}
      */
     build: function() {
-        return new FetchSet(this);
+        return new FetchSet(this, this.callback);
     }
 };
 
diff --git a/lib/commands/crdt/updatecounter.js b/lib/commands/crdt/updatecounter.js
index db51b57..80da2f7 100644
--- a/lib/commands/crdt/updatecounter.js
+++ b/lib/commands/crdt/updatecounter.js
@@ -46,18 +46,18 @@ var CounterOp = require('../../protobuf/riakprotobuf').getProtoFor('CounterOp');
  * @param {String} options.bucketType the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
  * @param {String} [options.key] the key for the counter you want to store.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Number} options.callback.response the counter value from Riak if returnbody is set.
  * @param {Number} [options.w] the W value to use.
  * @param {Number} [options.dw] the DW value to use.
  * @param {Number} [options.pw] the PW value to use.
  * @param {Boolean} [options.returnBody] return the counter.
  * @param {Number} [options.timeout] set a timeout for this command.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Number} callback.response the counter value from Riak if returnBody is set.
  * @extends CommandBase
  */
-function UpdateCounter(options) { 
-    CommandBase.call(this, 'DtUpdateReq', 'DtUpdateResp');
+function UpdateCounter(options, callback) { 
+    CommandBase.call(this, 'DtUpdateReq', 'DtUpdateResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -116,30 +116,19 @@ UpdateCounter.prototype.onSuccess = function(dtUpdateResp) {
         var value = dtUpdateResp.counter_value.toNumber();
         var response = { generatedKey: key, counterValue: value };
 
-        this.options.callback(null, response);
+        this._callback(null, response);
     } else {
-        this.options.callback(null, null);
+        this._callback(null, null);
     }
     
 };
 
-UpdateCounter.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-UpdateCounter.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-    
-
-
 var schema = Joi.object().keys({
     bucket: Joi.string().required(),
     bucketType: Joi.string().required(),
     key: Joi.string().default(null).optional(),
     increment: Joi.number().required(),
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
     w: Joi.number().default(null).optional(),
     dw: Joi.number().default(null).optional(),
     pw: Joi.number().default(null).optional(),
@@ -288,7 +277,7 @@ Builder.prototype = {
      * @return {UpdateCounter} a StoreValue instance
      */
     build : function() {
-        return new UpdateCounter(this);
+        return new UpdateCounter(this, this.callback);
     }
 };
 
diff --git a/lib/commands/crdt/updateset.js b/lib/commands/crdt/updateset.js
index 4526306..4bdd9c9 100644
--- a/lib/commands/crdt/updateset.js
+++ b/lib/commands/crdt/updateset.js
@@ -7,8 +7,8 @@ var Rpb = require('../../protobuf/riakprotobuf');
 var DtOp = Rpb.getProtoFor('DtOp');
 var SetOp = Rpb.getProtoFor('SetOp');
 
-function UpdateSet(options) {
-    CommandBase.call(this, 'DtUpdateReq', 'DtUpdateResp');
+function UpdateSet(options, callback) {
+    CommandBase.call(this, 'DtUpdateReq', 'DtUpdateResp', callback);
 
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -100,20 +100,12 @@ UpdateSet.prototype.onSuccess = function(dtUpdateResp) {
         response.value = stringlist(response.valueBuffers);
     }
 
-    this.options.callback(null, response);
+    this._callback(null, response);
     return true;
 };
 
-UpdateSet.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-
-UpdateSet.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
 
     //namespace
     bucket: Joi.string().required(),
@@ -165,7 +157,7 @@ Builder.prototype = {
      * @return {FetchSet}
      */
     build: function() {
-        return new UpdateSet(this);
+        return new UpdateSet(this, this.callback);
     }
 };
 
diff --git a/lib/commands/kv/deletevalue.js b/lib/commands/kv/deletevalue.js
index 553a72f..01f86b5 100644
--- a/lib/commands/kv/deletevalue.js
+++ b/lib/commands/kv/deletevalue.js
@@ -37,9 +37,6 @@ var Joi = require('joi');
  * @param {String} options.bucket the bucket in riak.
  * @param {String} options.key the key for the object you want to delete.
  * @param {Buffer} [options.vclock] the vector clock to use.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Boolean} options.callback.response the response from Riak
  * @param {Number} [options.timeout] set a timeout for this operation.
  * @param {Number} [options.r] the R value to use.
  * @param {Number} [options.pr] the PR value to use.
@@ -47,10 +44,13 @@ var Joi = require('joi');
  * @param {Number} [options.dw] the DW value to use.
  * @param {Number} [options.pw] the PW value to use.
  * @param {Number} [options.rw] the RW value to use.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Boolean} callback.response the response from Riak
  * @extends CommandBase
  */
-function DeleteValue(options) {
-    CommandBase.call(this, 'RpbDelReq', 'RpbDelResp');
+function DeleteValue(options, callback) {
+    CommandBase.call(this, 'RpbDelReq', 'RpbDelResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -116,19 +116,11 @@ DeleteValue.prototype.onSuccess = function(rpbGetResp) {
 
     // There is no body to a RpbDelResp. RpbDelReq either 
     // succeeds or returns an error. rpbDelResp will always be null
-    this.options.callback(null, true);
+    this._callback(null, true);
     return true;
 
 };
 
-DeleteValue.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-DeleteValue.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     bucket: Joi.string().required(),
     bucketType: Joi.string().default('default'),
@@ -140,7 +132,7 @@ var schema = Joi.object().keys({
     pw: Joi.number().optional(),
     rw: Joi.number().optional(),
     timeout: Joi.number().default(null),
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
     vclock: Joi.binary().optional()
 });
 
@@ -304,7 +296,7 @@ Builder.prototype = {
      * @return {DeleteValue} an instance of DeleteValue
      */
     build : function() {
-        return new DeleteValue(this);
+        return new DeleteValue(this, this.callback);
     }
 };
 
diff --git a/lib/commands/kv/fetchbucketprops.js b/lib/commands/kv/fetchbucketprops.js
index a9c3a5c..49e1a57 100644
--- a/lib/commands/kv/fetchbucketprops.js
+++ b/lib/commands/kv/fetchbucketprops.js
@@ -40,14 +40,14 @@ var Joi = require('joi');
  * @param {Object} options the options for this command
  * @param {String} [options.bucketType] the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Object} options.callback.response the response from Riak
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Object} callback.response the response from Riak
  * @extends CommandBase
  */
-function FetchBucketProps(options) {
+function FetchBucketProps(options, callback) {
     
-    CommandBase.call(this, 'RpbGetBucketReq','RpbGetBucketResp');
+    CommandBase.call(this, 'RpbGetBucketReq','RpbGetBucketResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -142,20 +142,11 @@ FetchBucketProps.prototype.onSuccess = function(rpbGetBucketResp) {
             fun: linkfun.function.toString('utf8') };
     }
     
-    this.options.callback(null, bucketProps);
+    this._callback(null, bucketProps);
     return true;
     
 };
 
-FetchBucketProps.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-FetchBucketProps.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 FetchBucketProps.prototype._parseHooks = function(rpbCommitHooks) {
     var hooks = new Array(rpbCommitHooks.length);
     for (var i = 0; i < rpbCommitHooks.length; i++) {
@@ -173,7 +164,7 @@ FetchBucketProps.prototype._parseHooks = function(rpbCommitHooks) {
 var schema = Joi.object().keys({
    bucket: Joi.string().required(),
    bucketType: Joi.string().default('default'),
-   callback: Joi.func().required()
+   callback: Joi.func().strip().optional()
 });
 
 /**
@@ -227,7 +218,7 @@ Builder.prototype = {
      * @return {GetBucketProps}
      */
     build : function() {
-        return new FetchBucketProps(this);
+        return new FetchBucketProps(this, this.callback);
     }
 };
 
diff --git a/lib/commands/kv/fetchvalue.js b/lib/commands/kv/fetchvalue.js
index 7402f9b..c6fdaf4 100644
--- a/lib/commands/kv/fetchvalue.js
+++ b/lib/commands/kv/fetchvalue.js
@@ -45,9 +45,6 @@ var Joi = require('joi');
  * @param {String} [options.bucketType] the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
  * @param {String} options.key the key for the object you want to fetch.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {FetchValue.Response} options.callback.response the response from Riak
  * @param {Boolean} options.convertToJs convert the values stored in riak to a JS object using JSON.parse() 
  * @param {Function} options.conflictResolver resolve sibling to a single RiakObject
  * @param {RiakObject[]|Object[]} options.conflictResolver.objects the array of objects returned from Riak.
@@ -59,12 +56,16 @@ var Joi = require('joi');
  * @param {Boolean} [options.returnDeletedVClock] true to return tombstones
  * @param {Boolean} [options.headOnly] Return only the metadata.
  * @param {Buffer} [options.ifNotModified] Do not return the object if the supplied vclock matches. 
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {FetchValue.Response} callback.response the response from Riak
+
  * @extends CommandBase
  * 
  */ 
-function FetchValue(options) {
+function FetchValue(options, callback) {
     
-    CommandBase.call(this, 'RpbGetReq', 'RpbGetResp');
+    CommandBase.call(this, 'RpbGetReq', 'RpbGetResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -125,7 +126,7 @@ FetchValue.prototype.onSuccess = function(rpbGetResp) {
     // a message code and zero bytes when that's the case. 
     // Because that makes sense!
     if (rpbGetResp === null) {
-        this.callback(null, new Response(true, false, []));
+        this._callback(null, new Response(true, false, []));
     } else {
 
         var pbContentArray = rpbGetResp.getContent();
@@ -143,7 +144,7 @@ FetchValue.prototype.onSuccess = function(rpbGetResp) {
             riakObject.key = this.key;
             riakObject.bucket = this.bucket;
             riakObject.bucketType = this.bucketType;
-            this.callback(null, new Response(false, false, [riakObject]));
+            this._callback(null, new Response(false, false, [riakObject]));
         } else {
 
             var values = new Array(pbContentArray.length);
@@ -159,9 +160,9 @@ FetchValue.prototype.onSuccess = function(rpbGetResp) {
                 values[i] = riakObject;
             }
             if (this.options.conflictResolver) {
-                this.options.callback(null, new Response(false, false, [this.options.conflictResolver(values)]));
+                this._callback(null, new Response(false, false, [this.options.conflictResolver(values)]));
             } else {
-                this.options.callback(null, new Response(false, false, values));
+                this._callback(null, new Response(false, false, values));
 
             }
         }
@@ -170,15 +171,6 @@ FetchValue.prototype.onSuccess = function(rpbGetResp) {
     return true;
 };
     
-FetchValue.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-FetchValue.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
    bucket: Joi.string().required(),
    bucketType: Joi.string().default('default'),
@@ -191,7 +183,7 @@ var schema = Joi.object().keys({
    headOnly: Joi.boolean().default(false),
    ifNotModified: Joi.binary().default(null),
    timeout: Joi.number().default(null),
-   callback: Joi.func().required(),
+   callback: Joi.func().strip().optional(),
    conflictResolver: Joi.func().optional(),
    convertToJs: Joi.boolean().default(false)
 });
@@ -380,7 +372,7 @@ Builder.prototype = {
      * @return {FetchValue}
      */
     build : function() {
-        return new FetchValue(this);
+        return new FetchValue(this, this.callback);
     }
         
 };
diff --git a/lib/commands/kv/listbuckets.js b/lib/commands/kv/listbuckets.js
index 4f07051..4d749be 100644
--- a/lib/commands/kv/listbuckets.js
+++ b/lib/commands/kv/listbuckets.js
@@ -40,16 +40,17 @@ var Joi = require('joi');
  * @constructor
  * @param {Object} options the options for this command
  * @param {String} [options.bucketType=default] the bucket type in riak.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {String[]} options.callback.response the buckets returned from riak
- * @param {Boolean} options.callback.done if you have received all the buckets
  * @param {Boolean} [stream=true] whether to stream or accumulate the result before calling callback
  * @param {Number} [options.timeout] set a timeout for this operation.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {String[]} callback.response the buckets returned from riak
+ * @param {Boolean} callback.done if you have received all the buckets
+
  * @extends CommandBase
  */
-function ListBuckets(options) {
-    CommandBase.call(this, 'RpbListBucketsReq', 'RpbListBucketsResp');
+function ListBuckets(options, callback) {
+    CommandBase.call(this, 'RpbListBucketsReq', 'RpbListBucketsResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -93,11 +94,11 @@ ListBuckets.prototype.onSuccess = function(rpbListBucketsResp) {
     }
         
     if (this.options.stream) {
-        this.options.callback(null, bucketsToSend, rpbListBucketsResp.done);
+        this._callback(null, bucketsToSend, rpbListBucketsResp.done);
     } else {
         Array.prototype.push.apply(this.buckets, bucketsToSend);
         if (rpbListBucketsResp.done) {
-            this.options.callback(null, this.buckets, rpbListBucketsResp.done);
+            this._callback(null, this.buckets, rpbListBucketsResp.done);
         }
     }
     
@@ -105,18 +106,9 @@ ListBuckets.prototype.onSuccess = function(rpbListBucketsResp) {
     
 };
 
-ListBuckets.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-ListBuckets.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
    bucketType: Joi.string().default('default'),
-   callback : Joi.func().required(),
+   callback : Joi.func().strip().optional(),
    stream : Joi.boolean().default(true),
    timeout: Joi.number().default(null)
 });
@@ -194,7 +186,7 @@ Builder.prototype = {
      * @return {ListKeys} 
      */
     build : function() {
-        return new ListBuckets(this);
+        return new ListBuckets(this, this.callback);
     }
 };
 
diff --git a/lib/commands/kv/listkeys.js b/lib/commands/kv/listkeys.js
index d7a27f6..7c3e30e 100644
--- a/lib/commands/kv/listkeys.js
+++ b/lib/commands/kv/listkeys.js
@@ -42,16 +42,17 @@ var Joi = require('joi');
  * @param {Object} options the options for this command
  * @param {String} [options.bucketType=default] the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {String[]} options.callback.response the keys returned from riak
- * @param {Boolean} options.callback.done if you have received all the keys
  * @param {Boolean} [stream=true] whether to stream or accumulate the result before calling callback
  * @param {Number} [options.timeout] set a timeout for this operation.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {String[]} callback.response the keys returned from riak
+ * @param {Boolean} callback.done if you have received all the keys
+
  * @extends CommandBase
  */
-function ListKeys(options) {
-    CommandBase.call(this, 'RpbListKeysReq', 'RpbListKeysResp');
+function ListKeys(options, callback) {
+    CommandBase.call(this, 'RpbListKeysReq', 'RpbListKeysResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -94,11 +95,11 @@ ListKeys.prototype.onSuccess = function(rpbListKeysResp) {
     }
         
     if (this.options.stream) {
-        this.options.callback(null, keysToSend, rpbListKeysResp.done);
+        this._callback(null, keysToSend, rpbListKeysResp.done);
     } else {
         Array.prototype.push.apply(this.keys, keysToSend);
         if (rpbListKeysResp.done) {
-            this.options.callback(null, this.keys, rpbListKeysResp.done);
+            this._callback(null, this.keys, rpbListKeysResp.done);
         }
     }
     
@@ -106,19 +107,10 @@ ListKeys.prototype.onSuccess = function(rpbListKeysResp) {
     
 };
 
-ListKeys.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-ListKeys.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
    bucket: Joi.string().required(),
    bucketType: Joi.string().default('default'),
-   callback : Joi.func().required(),
+   callback : Joi.func().strip().optional(),
    stream : Joi.boolean().default(true),
    timeout: Joi.number().default(null)
 });
@@ -207,7 +199,7 @@ Builder.prototype = {
      * @return {ListKeys} 
      */
     build : function() {
-        return new ListKeys(this);
+        return new ListKeys(this, this.callback);
     }
     
 };
diff --git a/lib/commands/kv/secondaryindexquery.js b/lib/commands/kv/secondaryindexquery.js
index 12487ff..92d0911 100644
--- a/lib/commands/kv/secondaryindexquery.js
+++ b/lib/commands/kv/secondaryindexquery.js
@@ -46,11 +46,6 @@ var RiakObject = require('./riakobject');
  * @param {Object} options options for this command.
  * @param {String} [options.bucketType] the bucket type in riak.
  * @param {String} options.bucket the bucket in riak.
- * @param {Function} options.callback the callback to be executed by the command.
- * @param {String} options.callback.err  An error message
- * @param {Object[]} options.callback.response object keys returned by the query, and optionally the index keys.
- * @param {Boolean} options.callback.done if you have received all the results.
- * @param {Buffer} options.callback.continuation if a continuation was returned
  * @param {String|Number} [options.indexKey] a single index key to query
  * @param {String|Number} [options.rangeStart] Starting key for a range query
  * @param {String|Number} [options.rangeEnd] Ending key for a range query
@@ -59,11 +54,16 @@ var RiakObject = require('./riakobject');
  * @param {Buffer} [options.continuation] returned from a previous query that set maxResults. used for pagination.
  * @param {Boolean} [options.stream=true] whether to stream or accumulate the result before calling callback.
  * @param {Number} [options.timeout] set a timeout for this operation.
- * @param {Boolean} [options.paginationSort=false] true to sort a non-paginated query. 
+ * @param {Boolean} [options.paginationSort=false] true to sort a non-paginated query.
+ * @param {Function} callback the callback to be executed by the command.
+ * @param {String} callback.err  An error message
+ * @param {Object[]} callback.response object keys returned by the query, and optionally the index keys.
+ * @param {Boolean} callback.done if you have received all the results.
+ * @param {Buffer} callback.continuation if a continuation was returned
  * @extends CommandBase
  */
-function SecondaryIndexQuery(options) {
-    CommandBase.call(this, 'RpbIndexReq', 'RpbIndexResp');
+function SecondaryIndexQuery(options, callback) {
+    CommandBase.call(this, 'RpbIndexReq', 'RpbIndexResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -165,11 +165,11 @@ SecondaryIndexQuery.prototype.onSuccess = function(rpbIndexResp) {
     var continuation = rpbIndexResp.continuation;
     
     if (this.options.stream) {
-        this.options.callback(null, resultsToReturn, rpbIndexResp.done, continuation );
+        this._callback(null, resultsToReturn, rpbIndexResp.done, continuation );
     } else {
         Array.prototype.push.apply(this.responses, resultsToReturn);
         if (rpbIndexResp.done) {
-            this.options.callback(null, this.responses, true, continuation);
+            this._callback(null, this.responses, true, continuation);
         }
     }
     
@@ -177,16 +177,6 @@ SecondaryIndexQuery.prototype.onSuccess = function(rpbIndexResp) {
     
 };
 
-SecondaryIndexQuery.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-SecondaryIndexQuery.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
-
 // Explicitly set null on options for protobufs and for conditionals
 var schema = Joi.object().keys({
    bucket: Joi.string().required(),
@@ -198,7 +188,7 @@ var schema = Joi.object().keys({
    returnKeyAndIndex: Joi.boolean().default(null).optional(),
    maxResults: Joi.number().default(null).optional(),
    continuation: Joi.any().default(null).optional(),
-   callback: Joi.func().required(),
+   callback: Joi.func().strip().optional(),
    timeout: Joi.number().default(null).optional(),
    stream: Joi.boolean().default(true).optional(),
    paginationSort: Joi.boolean().default(null).optional()
@@ -379,7 +369,7 @@ Builder.prototype = {
      * @return {SecondaryIndexQuery}
      */
     build : function() {
-        return new SecondaryIndexQuery(this);
+        return new SecondaryIndexQuery(this, this.callback);
     }
     
 };
diff --git a/lib/commands/kv/storebucketprops.js b/lib/commands/kv/storebucketprops.js
index 44cb8a4..c8359dc 100644
--- a/lib/commands/kv/storebucketprops.js
+++ b/lib/commands/kv/storebucketprops.js
@@ -41,9 +41,6 @@ var Joi = require('joi');
  * @param {Object} options the properties to store
  * @param {String} options.bucket the bucket in riak.
  * @param {String} [options.bucketType] the bucket type in riak.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {Boolean} options.callback.response the response from Riak
  * @param {Number} [options.r] the R value.
  * @param {Number} [options.pr] the PR value.
  * @param {Number} [options.w] the W value.
@@ -65,10 +62,13 @@ var Joi = require('joi');
  * @param {Object} [options.chashKeyfun] a object representing the Erlang func to use.
  * @param {Object[]} [options.precommit] array of precommit hooks
  * @param {Object[]} [options.postcommit] array of postcommit hooks
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {Boolean} callback.response the response from Riak
  * @extends CommandBase
  */
-function StoreBucketProps(options) {
-    CommandBase.call(this, 'RpbSetBucketReq', 'RpbSetBucketResp');
+function StoreBucketProps(options, callback) {
+    CommandBase.call(this, 'RpbSetBucketReq', 'RpbSetBucketResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -159,22 +159,11 @@ StoreBucketProps.prototype.onSuccess = function(rpbSetBucketResp) {
     
     // There is not response from riak except a code. rpbSetBucketResp
     // will be null upon success.
-    this.options.callback(null, true);
+    this._callback(null, true);
     return true;
     
 };
 
-StoreBucketProps.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-StoreBucketProps.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};    
-    
-
-
 var schema = Joi.object().keys({
     bucket: Joi.string().required(),
     bucketType: Joi.string().default('default'),
@@ -184,7 +173,7 @@ var schema = Joi.object().keys({
     dw: Joi.number().optional(),
     pw: Joi.number().optional(),
     rw: Joi.number().optional(),
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
     notFoundOk: Joi.boolean().optional(),
     nVal: Joi.number().optional(),
     allowMult: Joi.boolean().optional(),
@@ -494,7 +483,7 @@ Builder.prototype = {
      * @return {StoreBucketProps}
      */
     build : function() {
-        return new StoreBucketProps(this);
+        return new StoreBucketProps(this, this.callback);
     }
     
 };
diff --git a/lib/commands/kv/storevalue.js b/lib/commands/kv/storevalue.js
index 6bebbbb..f6dc726 100644
--- a/lib/commands/kv/storevalue.js
+++ b/lib/commands/kv/storevalue.js
@@ -44,9 +44,6 @@ var Joi = require('joi');
  * @param {String} options.bucket the bucket in riak.
  * @param {String} [options.key] the key for the object you want to store.
  * @param {Buffer} [options.vclock] the vector clock to use.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err An error message
- * @param {StoreValue.Response} options.callback.response the response from Riak
  * @param {RiakObject|String|Buffer|Object} options.value the value to store in Riak
  * @param {Number} [options.w] the W value to use.
  * @param {Number} [options.dw] the DW value to use.
@@ -56,10 +53,13 @@ var Joi = require('joi');
  * @param {Number} [options.timeout] set a timeout for this command.
  * @param {Boolean} [options.ifNotModified] the if_not_modified flag.
  * @param {Boolean} [options.ifNoneMatch] the if_none_match flag.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err An error message
+ * @param {StoreValue.Response} callback.response the response from Riak
  * @extends CommandBase
  */
-function StoreValue(options) {
-    CommandBase.call(this, 'RpbPutReq', 'RpbPutResp');
+function StoreValue(options, callback) {
+    CommandBase.call(this, 'RpbPutReq', 'RpbPutResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -182,28 +182,19 @@ StoreValue.prototype.onSuccess = function(rpbPutResp) {
             values =[];
         }
 
-        this.options.callback(null, new Response(values, responseKey));
+        this._callback(null, new Response(values, responseKey));
     } else {
-        this.options.callback(null, new Response([]));
+        this._callback(null, new Response([]));
     }
     
     return true;    
 };
 
-StoreValue.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-StoreValue.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     bucket: Joi.string().required(),
     bucketType: Joi.string().default('default'),
     key: Joi.string().optional(),
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
     value: Joi.any().required(),
     w: Joi.number().optional(),
     dw: Joi.number().optional(),
@@ -439,7 +430,7 @@ Builder.prototype = {
      * @return {StoreValue} a StoreValue instance
      */
     build : function() {
-        return new StoreValue(this);
+        return new StoreValue(this, this.callback);
     }
     
 };
diff --git a/lib/commands/mapreduce/mapreduce.js b/lib/commands/mapreduce/mapreduce.js
index 1b7fbe3..f977d24 100644
--- a/lib/commands/mapreduce/mapreduce.js
+++ b/lib/commands/mapreduce/mapreduce.js
@@ -64,17 +64,16 @@ var Joi = require('joi');
  * @extends CommandBase
  */
 function MapReduce(query, callback, stream) {
-    CommandBase.call(this, 'RpbMapRedReq', 'RpbMapRedResp');
+    CommandBase.call(this, 'RpbMapRedReq', 'RpbMapRedResp', callback);
     
     var self = this;
-    Joi.validate({ query: query, callback: callback, stream: stream}, schema, function(err, options) {
+    Joi.validate({ query: query, stream: stream}, schema, function(err, options) {
        
         if (err) {
             throw err;
         }
     
         self.query = options.query;
-        self.callback = options.callback;
         self.stream = options.stream;
     });
     
@@ -119,29 +118,19 @@ MapReduce.prototype.onSuccess = function(rpbMapRedResp) {
     
     if (!this.stream) {
         if (done) {
-            this.callback(null, this.response);
+            this._callback(null, this.response, done);
         }
     } else {
-        this.callback(null, { phase : phase, response: responseArray, done: done });
+        this._callback(null, { phase : phase, response: responseArray}, done);
     }
     
     return done;
     
 };
 
-MapReduce.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-MapReduce.prototype.onError = function(msg) {
-    this.callback(msg, null);
-};
-
-
 var schema = Joi.object().keys({
     query: Joi.string().required(),
-    callback: Joi.func().required(),
+    callback: Joi.func().strip().optional(),
     stream: Joi.boolean().default(true).optional()
 });
 
diff --git a/lib/commands/ping.js b/lib/commands/ping.js
index 8e512e9..d015e8e 100644
--- a/lib/commands/ping.js
+++ b/lib/commands/ping.js
@@ -32,8 +32,7 @@ var logger = require('winston');
  * @extends CommandBase
  */ 
 function Ping(callback) {
-    CommandBase.call(this, 'RpbPingReq', 'RpbPingResp');
-    this.callback = callback;
+    CommandBase.call(this, 'RpbPingReq', 'RpbPingResp', callback);
 }
 
 inherits(Ping, CommandBase);
@@ -47,18 +46,9 @@ Ping.prototype.constructPbRequest = function() {
 
 Ping.prototype.onSuccess = function(rpbPingResp) {
     logger.debug("Ping#onSuccess");
-    this.callback(null, true);
+    this._callback(null, true);
     return true;
 };
     
-Ping.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-Ping.prototype.onError = function(msg) {
-    this.callback(msg, null);
-};
-
 module.exports = Ping;
 
diff --git a/lib/commands/yokozuna/deleteindex.js b/lib/commands/yokozuna/deleteindex.js
index 94bf1c5..df0828e 100644
--- a/lib/commands/yokozuna/deleteindex.js
+++ b/lib/commands/yokozuna/deleteindex.js
@@ -41,14 +41,14 @@ var Joi = require('joi');
  * @constructor
  * @param {options} the options for this command
  * @param {String} [options.indexName] the name of the index to delete
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err error message
- * @param {Boolean} options.callback.response operation either succeeds or returns error. This will be true.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err error message
+ * @param {Boolean} callback.response operation either succeeds or returns error. This will be true.
  * @extends CommandBase
  */
-function DeleteIndex(options) {
+function DeleteIndex(options, callback) {
     
-    CommandBase.call(this, 'RpbYokozunaIndexDeleteReq' , 'RpbDelResp');
+    CommandBase.call(this, 'RpbYokozunaIndexDeleteReq' , 'RpbDelResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -80,22 +80,13 @@ DeleteIndex.prototype.onSuccess = function(RpbDelResp) {
     
     // RpbDelResp is simply null (no body)
     
-    this.options.callback(null, true);
+    this._callback(null, true);
     return true;
 };
 
-DeleteIndex.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-DeleteIndex.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     indexName: Joi.string().required(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -144,7 +135,7 @@ Builder.prototype = {
      * @return {DeleteIndex}
      */
     build : function() {
-        return new DeleteIndex(this);
+        return new DeleteIndex(this, this.callback);
     }
 };
 
diff --git a/lib/commands/yokozuna/fetchindex.js b/lib/commands/yokozuna/fetchindex.js
index aa86981..78d48c6 100644
--- a/lib/commands/yokozuna/fetchindex.js
+++ b/lib/commands/yokozuna/fetchindex.js
@@ -41,14 +41,14 @@ var Joi = require('joi');
  * @constructor
  * @param {options} the options for this command
  * @param {String} [options.indexName] the name of a specific index to fetch. If not supplied, all are returned.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err error message
- * @param {Object[]} options.callback.response array of indexes.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err error message
+ * @param {Object[]} callback.response array of indexes.
  * @extends CommandBase
  */
-function FetchIndex(options) {
+function FetchIndex(options, callback) {
     
-    CommandBase.call(this, 'RpbYokozunaIndexGetReq' , 'RpbYokozunaIndexGetResp');
+    CommandBase.call(this, 'RpbYokozunaIndexGetReq' , 'RpbYokozunaIndexGetResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -94,22 +94,14 @@ FetchIndex.prototype.onSuccess = function(rpbYokozunaIndexGetResp) {
     } else {
         indexesToReturn = [];
     }
-    this.options.callback(null, indexesToReturn);
+    this._callback(null, indexesToReturn);
     return true;
 };
 
-FetchIndex.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-FetchIndex.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
 
 var schema = Joi.object().keys({
     indexName: Joi.string().default(null).optional(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -158,7 +150,7 @@ Builder.prototype = {
      * @return {FetchIndex}
      */
     build : function() {
-        return new FetchIndex(this);
+        return new FetchIndex(this, this.callback);
     }
 };
 
diff --git a/lib/commands/yokozuna/fetchschema.js b/lib/commands/yokozuna/fetchschema.js
index bf611c1..a83944c 100644
--- a/lib/commands/yokozuna/fetchschema.js
+++ b/lib/commands/yokozuna/fetchschema.js
@@ -41,14 +41,14 @@ var Joi = require('joi');
  * @constructor
  * @param {options} the options for this command
  * @param {String} options.schemaName the name of the schema to fetch.
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err error message
- * @param {Object[]} options.callback.response the schema
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err error message
+ * @param {Object[]} callback.response the schema
  * @extends CommandBase
  */
-function FetchSchema(options) {
+function FetchSchema(options, callback) {
     
-    CommandBase.call(this, 'RpbYokozunaSchemaGetReq' , 'RpbYokozunaSchemaGetResp');
+    CommandBase.call(this, 'RpbYokozunaSchemaGetReq' , 'RpbYokozunaSchemaGetResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -85,22 +85,13 @@ FetchSchema.prototype.onSuccess = function(rpbYokozunaSchemaGetResp) {
     }
     
     
-    this.options.callback(null, schema);
+    this._callback(null, schema);
     return true;
 };
 
-FetchSchema.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-FetchSchema.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     schemaName: Joi.string().default(null).optional(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -148,7 +139,7 @@ Builder.prototype = {
      * @return {FetchSchema}
      */
     build : function() {
-        return new FetchSchema(this);
+        return new FetchSchema(this, this.callback);
     }
 };
 
diff --git a/lib/commands/yokozuna/search.js b/lib/commands/yokozuna/search.js
index 50a921c..d24d4c2 100644
--- a/lib/commands/yokozuna/search.js
+++ b/lib/commands/yokozuna/search.js
@@ -52,10 +52,12 @@ var Joi = require('joi');
  * @param {String[]} [options.returnFields] Only return the provided fields.
  * @param {String} [options.presort] Sorts all of the results. Either "key" or "score".
  * @param {Function} callback the callback to execute when the comman completes.
+ * @param {String} callback.err an error message
+ * @param {Object} response the response from Riak (Solr)
  * @extends CommandBase
  */
-function Search(options)  {
-    CommandBase.call(this, 'RpbSearchQueryReq', 'RpbSearchQueryResp');
+function Search(options, callback)  {
+    CommandBase.call(this, 'RpbSearchQueryReq', 'RpbSearchQueryResp', callback);
     
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -139,22 +141,11 @@ Search.prototype.onSuccess = function(rpbSearchQueryResp) {
                    maxScore : rpbSearchQueryResp.max_score,
                    docs: docsToReturn };
                
-    this.options.callback(null, result);
+    this._callback(null, result);
     return true;
 
 };
 
-Search.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-Search.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
-
-
 var schema = Joi.object().keys({
     indexName: Joi.string().required(),
     q: Joi.string().required(),
@@ -166,7 +157,7 @@ var schema = Joi.object().keys({
     defaultOperation: Joi.string().default(null).optional(),
     returnFields: Joi.array().default([]).optional(),
     presort: Joi.string().default(null).optional(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -317,7 +308,7 @@ Builder.prototype = {
     * @return {Search}
     */
    build : function() {
-       return new Search(this);
+       return new Search(this, this.callback);
    }
     
 };
diff --git a/lib/commands/yokozuna/storeindex.js b/lib/commands/yokozuna/storeindex.js
index ad3619d..ed80593 100644
--- a/lib/commands/yokozuna/storeindex.js
+++ b/lib/commands/yokozuna/storeindex.js
@@ -43,14 +43,14 @@ var RpbYokozunaIndex = require('../../protobuf/riakprotobuf').getProtoFor('RpbYo
  * @param {options} the options for this command
  * @param {String} options.indexName the name of the index. 
  * @param (String) [options.schemaName=_yz_default] the name of the schema of this index. 
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err error message
- * @param {Boolean} options.callback.response the operation either succeeds or errors. This will be true.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err error message
+ * @param {Boolean} callback.response the operation either succeeds or errors. This will be true.
  * @extends CommandBase
  */
-function StoreIndex(options) {
+function StoreIndex(options, callback) {
     
-    CommandBase.call(this, 'RpbYokozunaIndexPutReq' , 'RpbPutResp');
+    CommandBase.call(this, 'RpbYokozunaIndexPutReq' , 'RpbPutResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -86,24 +86,15 @@ StoreIndex.prototype.onSuccess = function(RpbPutResp) {
     
     // RpbPutResp is simply null (no body)
     
-    this.options.callback(null, true);
+    this._callback(null, true);
     return true;
 };
 
 
-StoreIndex.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-StoreIndex.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     indexName: Joi.string().required(),
     schemaName: Joi.string().default(null).optional(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -163,7 +154,7 @@ Builder.prototype = {
      * @return {StoreIndex}
      */
     build : function() {
-        return new StoreIndex(this);
+        return new StoreIndex(this, this.callback);
     }
 };
 
diff --git a/lib/commands/yokozuna/storeschema.js b/lib/commands/yokozuna/storeschema.js
index bcbdd19..26fc0bc 100644
--- a/lib/commands/yokozuna/storeschema.js
+++ b/lib/commands/yokozuna/storeschema.js
@@ -43,14 +43,14 @@ var RpbYokozunaSchema = require('../../protobuf/riakprotobuf').getProtoFor('RpbY
  * @param {options} the options for this command
  * @param {String} options.schemaName the name of the schema to store.
  * @param {String} options.schema The XML that defines this schema
- * @param {Function} options.callback the callback to be executed when the operation completes.
- * @param {String} options.callback.err error message
- * @param {Boolean} options.callback.response the schemathis operation either succeeds or errors. This will be true.
+ * @param {Function} callback the callback to be executed when the operation completes.
+ * @param {String} callback.err error message
+ * @param {Boolean} callback.response the schemathis operation either succeeds or errors. This will be true.
  * @extends CommandBase
  */
-function StoreSchema(options) {
+function StoreSchema(options, callback) {
     
-    CommandBase.call(this, 'RpbYokozunaSchemaPutReq' , 'RpbPutResp');
+    CommandBase.call(this, 'RpbYokozunaSchemaPutReq' , 'RpbPutResp', callback);
     var self = this;
     Joi.validate(options, schema, function(err, options) {
        
@@ -83,23 +83,14 @@ StoreSchema.prototype.constructPbRequest = function() {
 StoreSchema.prototype.onSuccess = function(rpbPutResp) {
     
     // rpbPutResp will be null (no body)
-    this.options.callback(null, true);
+    this._callback(null, true);
     return true;
 };
 
-StoreSchema.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-StoreSchema.prototype.onError = function(msg) {
-    this.options.callback(msg, null);
-};
-
 var schema = Joi.object().keys({
     schemaName: Joi.string().required(),
     schema: Joi.string().required(),
-    callback: Joi.func().required()
+    callback: Joi.func().strip().optional()
 });
 
 /**
@@ -159,7 +150,7 @@ Builder.prototype = {
      * @return {StoreSchema}
      */
     build : function() {
-        return new StoreSchema(this);
+        return new StoreSchema(this, this.callback);
     }
 };
 
diff --git a/lib/core/authreq.js b/lib/core/authreq.js
index c40dd6b..0bc42a3 100644
--- a/lib/core/authreq.js
+++ b/lib/core/authreq.js
@@ -32,12 +32,12 @@ var logger = require('winston');
  * @param {Object} options
  * @param {String} options.user the user with which to authenticate (required)
  * @param {String} options.password the password with which to authenticate (optional)
- * @param {Function} options.callback the function to call when the command completes or errors
+ * @param {Function} callback the function to call when the command completes or errors
  * @extends CommandBase
  */ 
-function AuthReq(options) {
+function AuthReq(options, callback) {
 
-    CommandBase.call(this, 'RpbAuthReq', 'RpbAuthResp');
+    CommandBase.call(this, 'RpbAuthReq', 'RpbAuthResp', callback);
 
     var self = this;
     Joi.validate(options, schema, function(err, options) {
@@ -46,9 +46,6 @@ function AuthReq(options) {
         }
         self.user = options.user;
         self.password = options.password;
-        if (options.callback) {
-            self.callback = options.callback;
-        }
     });
 }
 
@@ -72,21 +69,10 @@ AuthReq.prototype.onSuccess = function(rpbAuthResp) {
     return true;
 };
     
-AuthReq.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-AuthReq.prototype.onError = function(msg) {
-    if (this.callback) {
-        this.callback(msg, null);
-    }
-};
-
 var schema = Joi.object().keys({
     user: Joi.string().required(),
     password: Joi.string().optional().allow('').default(''),
-    callback: Joi.func().optional()
+    callback: Joi.func().strip().optional()
 });
 
 module.exports = AuthReq;
diff --git a/lib/core/riakconnection.js b/lib/core/riakconnection.js
index 0a09858..4b34efb 100644
--- a/lib/core/riakconnection.js
+++ b/lib/core/riakconnection.js
@@ -93,7 +93,7 @@ RiakConnection.prototype._connected = function() {
         logger.debug('RiakConnection:_connected:StartTls');
         this._boundReceiveStartTls = this._receiveStartTls.bind(this);
         this._connection.on('data', this._boundReceiveStartTls);
-        var command = new StartTls();
+        var command = new StartTls(function(){});
         this.execute(command);
     } else {
         this._connection.on('data', this._receiveData.bind(this));
@@ -149,7 +149,7 @@ RiakConnection.prototype._receiveStartTls = function(data) {
 
     // Execute AuthReq command
     this.inFlight = false;
-    var command = new AuthReq(auth_options);
+    var command = new AuthReq(auth_options, function(){});
     this.execute(command);
 };
 
diff --git a/lib/core/starttls.js b/lib/core/starttls.js
index 2b3f079..844ddf0 100644
--- a/lib/core/starttls.js
+++ b/lib/core/starttls.js
@@ -28,24 +28,12 @@ var Joi = require('joi');
  * 
  * @class StartTls
  * @constructor
- * @param {Object} options
- * @param {Function} options.callback the function to call when the command completes or errors
+ * @param {Function} callback the function to call when the command completes or errors
  * @extends CommandBase
  */ 
-function StartTls(options) {
-    CommandBase.call(this, 'RpbStartTls', 'RpbStartTls');
+function StartTls(callback) {
+    CommandBase.call(this, 'RpbStartTls', 'RpbStartTls', callback);
 
-    if (options) {
-        var self = this;
-        Joi.validate(options, schema, function(err, options) {
-            if (err) {
-                throw err;
-            }
-            if (options.callback) {
-                self.callback = options.callback;
-            }
-        });
-    }
 }
 
 inherits(StartTls, CommandBase);
@@ -66,20 +54,6 @@ StartTls.prototype.onSuccess = function(rpbStartTlsResp) {
     return true;
 };
     
-StartTls.prototype.onRiakError = function(rpbErrorResp) {
-    this.onError(rpbErrorResp.getErrmsg().toString('utf8'));
-};
-    
-    
-StartTls.prototype.onError = function(msg) {
-    if (this.callback) {
-        this.callback(msg, null);
-    }
-};
-
-var schema = Joi.object().keys({
-    callback: Joi.func().optional()
-});
 
 module.exports = StartTls;
 


Test patch:
diff --git a/test/unit/mapreduce/mapreduce.js b/test/unit/mapreduce/mapreduce.js
index 30155ea..be5f793 100644
--- a/test/unit/mapreduce/mapreduce.js
+++ b/test/unit/mapreduce/mapreduce.js
@@ -20,7 +20,7 @@ var RpbMapRedResp = require('../../../lib/protobuf/riakprotobuf').getProtoFor('R
 
 var assert = require('assert');
 
-describe('FetchValue', function() {
+describe('MapReduce', function() {
     describe('Build', function() {
         it('should build a RpbMapRedReq correctly', function(done) {