var util = require('util');
var url = require('url');
var AbstractLevelDOWN = require('abstract-leveldown').AbstractLevelDOWN;
var RiakIterator = require('./iterator');
var riakpbc = require('riakpbc');
var async = require('async');


function toKey(key) {
    return typeof key === 'string' ? key : JSON.stringify(key);
}

function RiakDOWN(location) {
    if (!(this instanceof RiakDOWN)) return new RiakDOWN(location);

    AbstractLevelDOWN.call(this, location);

    var parsed = url.parse(location);
    this._client = riakpbc.createClient({ host: parsed.hostname, port: parsed.port });
    this._bucket = parsed.path.split('/')[1];
}

util.inherits(RiakDOWN, AbstractLevelDOWN);

RiakDOWN.prototype._open = function (options, callback) {
    this._client.connect(callback);
};

RiakDOWN.prototype._close = function (callback) {
    this._client.disconnect();

    process.nextTick(function () {
        callback();
    });
};

RiakDOWN.prototype._put = function (key, value, options, callback) {
    var bucket = options.bucket || this._bucket;
    
    this._client.put({
        bucket: bucket,
        key: toKey(key),
        content: {
            value: value,
            content_type: 'application/octet-stream'
        }
    }, callback);
};

RiakDOWN.prototype._get = function (key, options, callback) {
    var bucket = options.bucket || this._bucket;

    this._client.get({ bucket: bucket, key: toKey(key) }, function (err, reply) {
        if (err || (!reply || !reply.content || !reply.content.length)) {
            return callback(new Error('NotFound'));
        }

        callback(null, options.asBuffer === false ? reply.content[0].value.toString() : reply.content[0].value);
    });
};

RiakDOWN.prototype._del = function (key, options, callback) {
    var bucket = options.bucket || this._bucket;

    this._client.del({ bucket: bucket, key: toKey(key) }, callback);
};

RiakDOWN.prototype._batch = function (array, options, callback) {
    var self = this;

    async.eachSeries(array, function (item, cb) {
        if (item.type === 'put') {
            self._put(toKey(item.key), item.value, options, cb);
        } else if (item.type === 'del') {
            self._del(toKey(item.key), options, cb);
        }
    }, callback);
};

RiakDOWN.prototype._iterator = function (options) {
    return new RiakIterator(this, options);
};

module.exports = RiakDOWN;

@@ -1192,16 +1192,40 @@
 nt_type:
+ options.content_type %7C%7C
  'applic
@@ -1245,32 +1245,64 @@
 tream'%0A        %7D
+,%0A        vclock: options.vclock
 %0A    %7D, callback
@@ -1362,32 +1362,53 @@
 ns, callback) %7B%0A
+    var self = this;%0A
     var bucket =
@@ -1431,32 +1431,52 @@
 %7C%7C this._bucket;
+%0A    var newoptions;
 %0A%0A    this._clie
@@ -1680,32 +1680,85 @@
      %7D%0A%0A        
+if (reply.content.length === 1) %7B%0A            return 
 callback(null, o
@@ -1846,16 +1846,665 @@
 value);%0A
+        %7D%0A%0A        newoptions = JSON.parse(JSON.stringify(options)); // ugly clone so we don't modify the original options%0A        newoptions.bucket = bucket;%0A        RiakDOWN._siblingResolver(key, reply.content, newoptions, function (err, resolved) %7B%0A            if (err) %7B%0A                return callback(err);%0A            %7D%0A%0A            if (typeof resolved !== 'string' && !Buffer.isBuffer(resolved)) %7B%0A                resolved = JSON.stringify(resolved);%0A            %7D%0A%0A            self._put(key, resolved, %7B bucket: bucket, vclock: reply.vclock %7D, function (err) %7B%0A                self._get(key, options, callback);%0A            %7D);%0A        %7D);%0A
     %7D);%0A
@@ -2669,24 +2669,48 @@
 : toKey(key)
+, vclock: options.vclock
  %7D, callback
@@ -3129,24 +3129,24 @@
 (options) %7B%0A
-
     return n
@@ -3181,16 +3181,133 @@
 s);%0A%7D;%0A%0A
+RiakDOWN._siblingResolver = function (key, siblings, options, callback) %7B%0A    callback(null, siblings%5B0%5D.value);%0A%7D;%0A%0A
 module.e

implement sibling resolver, allow vclock to be passed to put and del operations