refactor: migrate to pull-streams

BREAKING:

- Rename `datastore` to `blockstore` and `datastore-legacy` to `datastore`.
- Stores now need to adhere to the [interface-pull-blob-store](https://github.com/ipfs/interface-pull-blob-store) definition.

Author: dignifiedquire

.travis.yml

@@ -2,7 +2,7 @@ sudo: false
 language: node_js
 node_js:
   - 4
-  - 5
+  - stable
 
 # Make sure we have new NPM.
 before_install:

package.json

@@ -31,26 +31,23 @@
   "devDependencies": {
     "abstract-blob-store": "^3.2.0",
     "aegir": "^6.0.1",
-    "async": "^2.0.1",
     "buffer-loader": "^0.0.1",
     "chai": "^3.5.0",
-    "fs-blob-store": "^5.2.1",
-    "idb-plus-blob-store": "^1.1.2",
+    "fs-pull-blob-store": "^0.3.0",
+    "idb-pull-blob-store": "^0.3.0",
     "lodash": "^4.15.0",
     "ncp": "^2.0.0",
     "pre-commit": "^1.1.3",
-    "rimraf": "^2.5.4",
-    "run-series": "^1.1.4"
+    "rimraf": "^2.5.4"
   },
   "dependencies": {
     "babel-runtime": "^6.11.6",
-    "bl": "^1.1.2",
-    "concat-stream": "^1.5.1",
     "ipfs-block": "^0.3.0",
     "lock": "^0.1.2",
-    "lockfile": "^1.0.1",
     "multihashes": "^0.2.2",
-    "xtend": "^4.0.1"
+    "pull-stream": "^3.4.3",
+    "run-series": "^1.1.4",
+    "safe-buffer": "^5.0.1"
   },
   "license": "MIT",
   "contributors": [

src/index.js

@@ -1,53 +1,55 @@
 'use strict'
 
+const assert = require('assert')
+
 const stores = require('./stores')
 
-function Repo (repoPath, options) {
-  if (!(this instanceof Repo)) {
-    return new Repo(repoPath, options)
+module.exports = class Repo {
+  constructor (repoPath, options) {
+    assert.equal(typeof repoPath, 'string', 'missing repoPath')
+    assert(options, 'missing options')
+    assert(options.stores, 'missing options.stores')
+
+    this.path = repoPath
+
+    const blobStores = initializeBlobStores(options.stores)
+
+    const setup = (name, needsLocks, needsConfig) => {
+      const args = [repoPath, blobStores[name]]
+      if (needsLocks) {
+        args.push(this.locks)
+      }
+
+      if (needsConfig) {
+        args.push(this.config)
+      }
+
+      return stores[name].setUp.apply(stores[name], args)
+    }
+
+    this.locks = setup('locks')
+    this.version = setup('version', true)
+    this.config = setup('config', true)
+    this.keys = setup('keys', true, true)
+    this.blockstore = setup('blockstore', true)
+  }
+
+  exists (callback) {
+    this.version.exists(callback)
   }
-  if (!options) { throw new Error('missing options param') }
-  if (!options.stores) { throw new Error('missing options.stores param') }
-
-  // If options.stores is an abstract-blob-store instead of a map, use it for
-  // all stores.
-  if (options.stores.prototype && options.stores.prototype.createWriteStream) {
-    const store = options.stores
-    options.stores = {
+}
+
+function initializeBlobStores (store) {
+  if (store.constructor) {
+    return {
       keys: store,
       config: store,
-      datastore: store,
+      blockstore: store,
       logs: store,
       locks: store,
       version: store
     }
   }
 
-  this.path = repoPath
-
-  this.locks = stores
-                  .locks
-                  .setUp(repoPath, options.stores.locks)
-
-  this.exists = (callback) => {
-    this.version.exists(callback)
-  }
-
-  this.version = stores
-                   .version
-                   .setUp(repoPath, options.stores.version, this.locks)
-
-  this.config = stores
-                .config
-                .setUp(repoPath, options.stores.config, this.locks)
-
-  this.keys = stores
-                .keys
-                .setUp(repoPath, options.stores.keys, this.locks, this.config)
-
-  this.datastore = stores
-                .datastore
-                .setUp(repoPath, options.stores.datastore, this.locks)
+  return store
 }
-
-exports = module.exports = Repo

src/stores/blockstore.js

@@ -0,0 +1,100 @@
+'use strict'
+
+const Block = require('ipfs-block')
+const pull = require('pull-stream')
+const Lock = require('lock')
+
+const PREFIX_LENGTH = 8
+
+exports = module.exports
+
+function multihashToPath (multihash, extension) {
+  extension = extension || 'data'
+  const filename = `${multihash.toString('hex')}.${extension}`
+  const folder = filename.slice(0, PREFIX_LENGTH)
+  const path = folder + '/' + filename
+
+  return path
+}
+
+exports.setUp = (basePath, BlobStore, locks) => {
+  const store = new BlobStore(basePath + '/blocks')
+  const lock = new Lock()
+
+  return {
+    get (key, extension, cb) {
+      if (typeof extension === 'function') {
+        cb = extension
+        extension = 'data'
+      }
+
+      if (!key) {
+        return cb(new Error('Invalid key'))
+      }
+      const p = multihashToPath(key, extension)
+
+      pull(
+        store.read(p),
+        pull.collect((err, values) => {
+          if (err) {
+            return cb(err)
+          }
+
+          if (extension === 'data') {
+            extension = 'protobuf'
+          }
+
+          const data = Buffer.concat(values)
+
+          cb(null, new Block(data, extension))
+        })
+      )
+    },
+
+    put (block, cb) {
+      if (!block || !block.data) {
+        return cb(new Error('Invalid block'))
+      }
+
+      const key = multihashToPath(block.key, block.extension)
+
+      lock(key, (release) => pull(
+        pull.values([block.data]),
+        store.write(key, release((err) => {
+          if (err) {
+            return cb(err)
+          }
+          cb(null, {key})
+        }))
+      ))
+    },
+
+    has (key, extension, cb) {
+      if (typeof extension === 'function') {
+        cb = extension
+        extension = undefined
+      }
+
+      if (!key) {
+        return cb(new Error('Invalid key'))
+      }
+
+      const path = multihashToPath(key, extension)
+      store.exists(path, cb)
+    },
+
+    delete (key, extension, cb) {
+      if (typeof extension === 'function') {
+        cb = extension
+        extension = undefined
+      }
+
+      if (!key) {
+        return cb(new Error('Invalid key'))
+      }
+
+      const path = multihashToPath(key, extension)
+      store.remove(path, cb)
+    }
+  }
+}

src/stores/config.js

@@ -1,40 +1,51 @@
 'use strict'
 
-const bl = require('bl')
+const pull = require('pull-stream')
+const series = require('run-series')
 
 exports = module.exports
 
-exports.setUp = (basePath, blobStore, locks) => {
-  const store = blobStore(basePath)
+exports.setUp = (basePath, BlobStore, locks) => {
+  const store = new BlobStore(basePath)
+  const configFile = 'config'
+
   return {
-    get: (callback) => {
-      store
-        .createReadStream('config')
-        .pipe(bl((err, config) => {
+    get (callback) {
+      pull(
+        store.read(configFile),
+        pull.collect((err, values) => {
           if (err) {
             return callback(err)
           }
+
+          const config = Buffer.concat(values)
           let result
           try {
             result = JSON.parse(config.toString())
           } catch (err) {
             return callback(err)
           }
+
           callback(null, result)
-        }))
+        })
+      )
     },
 
-    set: (config, callback) => {
-      locks.lock((err) => {
-        if (err) {
-          return callback(err)
+    set (config, callback) {
+      series([
+        (cb) => locks.lock(cb),
+        (cb) => {
+          pull(
+            pull.values([
+              new Buffer(JSON.stringify(config, null, 2))
+            ]),
+            store.write(configFile, cb)
+          )
         }
-
-        store.createWriteStream('config')
-          .once('finish', () => {
-            locks.unlock(callback)
-          })
-          .end(JSON.stringify(config, null, 2))
+      ], (err) => {
+        locks.unlock((err2) => {
+          callback(err || err2)
+        })
       })
     }
   }