Merge pull request #5 from Level/add/decode-stream

add decode stream

README.md

@@ -43,6 +43,10 @@ var codec = new Codec(db.options);
 
   Decode `value` with given `opts`.
 
+### #createDecodeStream([opts])
+
+  Create a transform stream that decodes incoming key value pairs, passed from readable streams created by levelup's `create*Stream`.
+
 ### #keyAsBuffer([opts])
 
   Check whether `opts` and the global `opts` call for a binary key encoding.

index.js

@@ -1,4 +1,5 @@
 var encodings = require('./lib/encodings');
+var Decoder = require('./lib/decoder');
 
 module.exports = Codec;
 
@@ -72,6 +73,10 @@ Codec.prototype.encodeLtgt = function(ltgt){
   return ret;
 };
 
+Codec.prototype.createDecodeStream = function(opts){
+  return new Decoder(this, opts);
+};
+
 Codec.prototype.keyAsBuffer = function(opts){
   return this._keyEncoding(opts).buffer;
 };

lib/decoder.js

@@ -0,0 +1,39 @@
+var EncodingError = require('level-errors').EncodingError;
+var Transform = require('readable-stream').Transform;
+var inherits = require('util').inherits;
+
+module.exports = Decoder;
+inherits(Decoder, Transform);
+
+function Decoder(codec, opts){
+  Transform.call(this, { objectMode: true });
+
+  if (opts.keys && opts.values) {
+    this._decode = function(key, value){
+      return {
+        key: codec.decodeKey(key, opts),
+        value: codec.decodeValue(value, opts)
+      };
+    };
+  } else if (opts.keys) {
+    this._decode = function(key) {
+      return codec.decodeKey(key, opts);
+    }; 
+  } else if (opts.values) {
+    this._decode = function(_, value){
+      return codec.decodeValue(value, opts);
+    }
+  } else {
+    this._decode = function(){};
+  }
+}
+
+Decoder.prototype._transform = function(kv, _, done){
+  try {
+    var value = this._decode(kv.key, kv.value);
+  } catch (err) {
+    return done(new EncodingError(err));
+  }
+  done(null, value);
+};
+

package.json

@@ -9,5 +9,9 @@
   },
   "scripts": {
     "test": "make test"
+  },
+  "dependencies": {
+    "level-errors": "^1.0.3",
+    "readable-stream": "^1.0.33"
   }
 }

test/decoder.js

@@ -0,0 +1,58 @@
+var test = require('tape');
+var Codec = require('..');
+var PassThrough = require('readable-stream').PassThrough;
+
+test('decoder', function(t){
+  var codec = new Codec({ keyEncoding: 'hex' });
+
+  t.test('keys and values', function(t){
+    var decoder = codec.createDecodeStream({
+      valueEncoding: 'json',
+      keys: true,
+      values: true
+    });
+    decoder.on('data', function(kv){
+      t.deepEqual(kv, {
+        key: '686579',
+        value: 'you'
+      });
+      t.end();
+    });
+
+    var source = PassThrough({ objectMode: true });
+    source.push({ key: new Buffer('hey'), value: '"you"' });
+
+    source.pipe(decoder);
+  });
+
+  t.test('keys', function(t){
+    var decoder = codec.createDecodeStream({
+      keys: true
+    });
+    decoder.on('data', function(key){
+      t.equal(key, '686579');
+      t.end();
+    });
+
+    var source = PassThrough({ objectMode: true });
+    source.push({ key: new Buffer('hey') });
+
+    source.pipe(decoder);
+  });
+
+  t.test('values', function(t){
+    var decoder = codec.createDecodeStream({
+      valueEncoding: 'hex',
+      values: true
+    });
+    decoder.on('data', function(value){
+      t.equal(value, '686579');
+      t.end();
+    });
+
+    var source = PassThrough({ objectMode: true });
+    source.push({ value: new Buffer('hey') });
+
+    source.pipe(decoder);
+  });
+});