refactor: Migrate to pull-streams

Author: dignifiedquire

README.md

@@ -64,10 +64,10 @@ const BlockService = require('ipfs-block-service')
 const BlockService = require('ipfs-block-service')
 const Block = require('ipfs-block')
 const IPFSRepo = require('ipfs-repo')  // storage repo
-const memstore = require('abstract-blob-store')  // in-memory store
+const Store = require(interface-pull-blob-store')  // in-memory store
 
 // setup a repo
-var repo = new IPFSRepo('example', { stores: memstore })
+var repo = new IPFSRepo('example', { stores: Store })
 
 // create a block
 const block = new Block('hello warld')

package.json

@@ -37,22 +37,21 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs-block-service#readme",
   "devDependencies": {
-    "aegir": "^3.0.0",
-    "bs58": "^3.0.0",
+    "aegir": "^6.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.4.0",
     "ipfs-block": "^0.3.0",
-    "ipfs-repo": "^0.7.1",
-    "lodash": "^4.8.2",
+    "ipfs-repo": "^0.8.0",
+    "lodash": "^4.15.0",
     "ncp": "^2.0.0",
-    "pre-commit": "^1.1.2",
-    "rimraf": "^2.5.1",
+    "pre-commit": "^1.1.3",
+    "rimraf": "^2.5.4",
     "run-series": "^1.1.4"
   },
   "dependencies": {
-    "multihashes": "^0.2.2",
+    "pull-stream": "^3.4.3",
     "run-parallel-limit": "^1.0.3"
   },
   "contributors": [

src/index.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const parallelLimit = require('run-parallel-limit')
-const mh = require('multihashes')
+const pull = require('pull-stream')
 
 // BlockService is a hybrid block datastore. It stores data in a local
 // datastore and may retrieve data from a remote Exchange.
@@ -24,72 +24,54 @@ module.exports = class BlockService {
     return this._bitswap != null
   }
 
-  addBlock (block, extension, callback) {
-    if (this.isOnline()) {
-      if (typeof extension === 'function') {
-        callback = extension
-        extension = undefined
-      }
-
-      this._bitswap.hasBlock(block, callback)
-    } else {
-      this._repo.datastore.put(block, extension, callback)
-    }
-  }
-
-  addBlocks (blocks, callback) {
-    if (!Array.isArray(blocks)) {
-      return callback(new Error('expects an array of Blocks'))
+  put (block, callback) {
+    callback = callback || (() => {})
+    if (!block) {
+      return callback(new Error('Missing block'))
     }
 
-    parallelLimit(blocks.map((block) => (next) => {
-      this.addBlock(block, next)
-    }), 100, callback)
+    pull(
+      pull.values([block]),
+      this.putStream(),
+      pull.onEnd(callback)
+    )
   }
 
-  getBlock (key, extension, callback) {
+  putStream () {
     if (this.isOnline()) {
-      if (typeof extension === 'function') {
-        callback = extension
-        extension = undefined
-      }
-
-      this._bitswap.getBlock(key, callback)
-    } else {
-      this._repo.datastore.get(key, extension, callback)
+      return this._bitswap.putStream()
     }
+
+    return this._repo.blockstore.putStream()
   }
 
-  getBlocks (multihashes, extension, callback) {
+  get (key, extension, callback) {
     if (typeof extension === 'function') {
       callback = extension
       extension = undefined
     }
 
-    if (!Array.isArray(multihashes)) {
-      return callback(new Error('Invalid batch of multihashes'))
-    }
+    callback = callback || (() => {})
 
-    if (this.isOnline()) {
-      this._bitswap.getBlocks(multihashes, (results) => {
-        callback(null, results)
+    pull(
+      this.getStream(key, extension),
+      pull.collect((err, result) => {
+        if (err) return callback(err)
+        callback(null, result[0])
       })
-      return
+    )
+  }
+
+  getStream (key, extension) {
+    if (this.isOnline()) {
+      return this._bitswap.getStream(key)
     }
 
-    const results = {}
-    parallelLimit(multihashes.map((key) => (next) => {
-      this._repo.datastore.get(key, extension, (error, block) => {
-        results[mh.toB58String(key)] = {error, block}
-        next()
-      })
-    }), 100, (err) => {
-      callback(err, results)
-    })
+    return this._repo.blockstore.getStream(key, extension)
   }
 
   deleteBlock (key, extension, callback) {
-    this._repo.datastore.delete(key, extension, callback)
+    this._repo.blockstore.delete(key, extension, callback)
   }
 
   deleteBlocks (multihashes, extension, callback) {