Reorangized and implementation of a basic dag store

index.js

@@ -1,67 +1,17 @@
 var ipfs = require('ipfs-api')
-var stream = require('stream')
-var multihash = require('multihashes')
-var base58 = require('base58-native')
+var block = require('./lib/block')
+var dag = require('./lib/dag')
 
 module.exports = function (opts) {
   if (!opts) opts = {}
 
   opts.host = opts.host || '127.0.0.1'
   opts.port = opts.port || 5001
 
-  var store = {}
-
   var node = ipfs(opts.host, opts.port)
 
-  store.createWriteStream = function (opts, cb) {
-    if (typeof opts === 'string') opts = {key: opts}
-    if (typeof opts === 'function') return store.createWriteStream(null, opts)
-
-    var bufferStream = new stream.PassThrough()
-
-    node.block.put(bufferStream, function (err, data) {
-      if (err) return cb(err)
-
-      cb(null, {
-        key: data.Key,
-        size: data.Size
-      })
-    })
-
-    return bufferStream
-  }
-
-  store.createReadStream = function (opts) {
-    if (typeof opts === 'string') opts = {key: opts}
-    var bufferStream = new stream.PassThrough()
-    node.block.get(opts.key, function (err, stream) {
-      if (err && !stream) return bufferStream.emit('error', 'No stream')
-      stream.pipe(bufferStream)
-    })
-
-    return bufferStream
+  return {
+    block: block(node),
+    dag: dag(node)
   }
-
-  store.exists = function (opts, cb) {
-    if (typeof opts === 'string') opts = {key: opts}
-    var buf
-
-    try {
-      buf = base58.decode(opts.key)
-      void multihash.decode(buf)
-    } catch (e) {
-      return cb(null, false)
-    }
-
-    node.block.get(opts.key, function (err, stream) {
-      if (err) return cb(err)
-      cb(null, true)
-    })
-  }
-
-  store.remove = function (opts, cb) {
-    cb(null, false)
-  }
-
-  return store
 }

lib/block.js

@@ -0,0 +1,61 @@
+var stream = require('stream')
+var multihash = require('multihashes')
+var base58 = require('base58-native')
+
+module.exports = function(node) {
+  var store = {}
+
+  store.createWriteStream = function (opts, cb) {
+    if (typeof opts === 'string') opts = {key: opts}
+    if (typeof opts === 'function') return store.createWriteStream(null, opts)
+
+    var bufferStream = new stream.PassThrough()
+
+    node.block.put(bufferStream, function (err, data) {
+      if (err) return cb(err)
+
+      cb(null, {
+        key: data.Key,
+        size: data.Size
+      })
+    })
+
+    return bufferStream
+  }
+
+  store.createReadStream = function (opts) {
+    if (typeof opts === 'string') opts = {key: opts}
+
+    var bufferStream = new stream.PassThrough()
+
+    node.block.get(opts.key, function (err, stream) {
+      if (err && !stream) return bufferStream.emit('error', err)
+      stream.pipe(bufferStream)
+    })
+
+    return bufferStream
+  }
+
+  store.exists = function (opts, cb) {
+    if (typeof opts === 'string') opts = {key: opts}
+    var buf
+
+    try {
+      buf = base58.decode(opts.key)
+      void multihash.decode(buf)
+    } catch (e) {
+      return cb(null, false)
+    }
+
+    node.block.get(opts.key, function (err, stream) {
+      if (err) return cb(err)
+      cb(null, true)
+    })
+  }
+
+  store.remove = function (opts, cb) {
+    cb(null, false)
+  }
+
+  return store;
+}

lib/dag.js

@@ -0,0 +1,140 @@
+var stream = require('stream')
+var multihash = require('multihashes')
+var base58 = require('base58-native')
+var merkledag = require('node-ipfs-mdag')
+var async = require('async')
+
+var Node = function() {
+  this.Links = []
+  this.Data = new Buffer('')
+}
+
+var Link = function() {
+  this.Hash = ''
+  this.Size = 0
+  this.Name = ''
+}
+
+module.exports = function(node) {
+  var store = {}
+  var root = new Node()
+  var root_hash = ''
+  var map = {}
+
+  store.createWriteStream = function (opts, cb) {
+    if (typeof opts === 'string') opts = {key: opts}
+    if (typeof opts === 'function') return store.createWriteStream(null, opts)
+    if (typeof opts === 'undefined' || opts === null) opts = {}
+
+    var bufferStream = new stream.PassThrough()
+    var buffer
+
+    bufferStream.on('data', function(chunk) {
+      if(!buffer) buffer = chunk
+      else {
+        buffer = Buffer.concat([buffer,chunk], buffer.length + chunk.length)
+      }
+    })
+
+    bufferStream.on('end', function() {
+      blob = new Node()
+      blob.Data = buffer.toString()
+      async.waterfall([
+        function (callback) {
+          var raw = new Buffer(JSON.stringify(blob))
+          node.object.put(raw, 'json', callback)
+        }, 
+        function (res, callback) {
+          node.object.stat(res.Hash, function(err, stat) {
+            callback(err, res, stat)
+          })
+        },
+        function (res, stat, callback) {
+          var link = new Link()
+
+          link.Hash = res.Hash
+          link.Size = stat.CumulativeSize
+          link.Name = opts.key || res.Hash
+
+          root.Links = root.Links.filter(function(item) {
+            return item.Name != opts.key
+          })
+
+          root.Links.push(link)
+
+          root.Data = root_hash
+          var raw = new Buffer(JSON.stringify(root))
+          node.object.put(raw, 'json', function(err, new_root) {
+            callback(err, link, new_root)
+          })
+        }, function (link, new_root, callback) {
+          map[opts.key] = true
+
+          callback(null, {
+            key: link.Name,
+            root: new_root.Hash,
+            hash: link.Hash,
+            size: link.Size
+          })
+
+          root_hash = new_root.Hash
+        }
+      ], function(err, meta) {
+        if (err) return cb(err)
+        cb(null, meta)
+      })
+    })
+
+    return bufferStream
+  }
+
+  store.createReadStream = function (opts) {
+    if (typeof opts === 'string') opts = {key: opts}
+
+    var bufferStream = new stream.PassThrough()
+    node.object.data((opts.root || root_hash) + '/' + opts.key, function (err, stream) {
+      if (err && !stream) return bufferStream.emit('error', err)
+      stream.pipe(bufferStream)
+    })
+
+    return bufferStream
+  }
+
+  store.exists = function (opts, cb) {
+    if (typeof opts === 'string') opts = {key: opts}
+    if(!root_hash) return cb(null, false)
+    node.object.stat((opts.root || root_hash) + '/' + opts.key, function(err, stat) {
+      if(err) cb(null, false)
+      else cb(null, true)
+    })
+  }
+
+  store.remove = function (opts, cb) {
+    async.waterfall([
+      function (callback) {
+
+        root.Links = root.Links.filter(function(item) {
+          return item.Name != opts.key
+        })
+
+        root.Data = root_hash
+        var raw = new Buffer(JSON.stringify(root))
+        node.object.put(raw, 'json', function(err, new_root) {
+          callback(err, new_root)
+        })
+      },
+      function (new_root, callback) {
+        map[opts.key] = false
+
+        callback(null, true)
+
+        root_hash = new_root.Hash
+      }
+    ], function(err, meta) {
+      if (err) return cb(err, false)
+      cb(null, meta)
+    })
+  }
+
+  return store;
+}

package.json

@@ -33,10 +33,9 @@
     "lint"
   ],
   "dependencies": {
+    "async": "^1.2.0",
     "base58-native": "^0.1.4",
-    "end-of-stream": "^1.1.0",
     "ipfs-api": "^1.1.7",
     "multihashes": "^0.1.2",
-    "vinyl": "^0.4.6"
   }
 }

tests/block.js

@@ -0,0 +1,15 @@
+var test = require('tape')
+var abstractBlobTests = require('abstract-blob-store/tests')
+var blobs = require('./')
+
+var common = {
+  setup: function (t, cb) {
+    var store = blobs().block
+    cb(null, store)
+  },
+  teardown: function (t, store, blob, cb) {
+    cb()
+  }
+}
+
+abstractBlobTests(test, common)

test.js → tests/dag.js

@@ -4,7 +4,7 @@ var blobs = require('./')
 
 var common = {
   setup: function (t, cb) {
-    var store = blobs()
+    var store = blobs().dag
     cb(null, store)
   },
   teardown: function (t, store, blob, cb) {