fix(block-service): use correct cb interface for writes

fixes #13

src/block-service.js

@@ -9,38 +9,9 @@ const async = require('async')
 // It uses an internal `datastore.Datastore` instance to store values.
 function BlockService (ipfsRepo, exchange) {
   this.addBlock = (block, callback) => {
-    const ws = ipfsRepo.datastore.createWriteStream(block.key, block.extension)
-
-    let done = false
+    const ws = ipfsRepo.datastore.createWriteStream(block.key, block.extension, callback)
 
     ws.write(block.data)
-
-    ws.once('error', (err) => {
-      done = true
-      callback(err)
-    })
-
-    ws.once('finish', () => {
-      if (!done) {
-        // Important to note: Writing to a stream
-        // isn't an atomic process, because streams can be
-        // piped, and the finish of one only represents that
-        // the data was buffered to the next one.
-        // This is something known and 'accepted' on the
-        // streams API, however, since we expose a callback
-        // interface on BlockService and a streams one,
-        // the users will expect for the callback to be fired
-        // when the final write was concluded. We add a
-        // timeout to ensure that.
-        // TODO: Create an elegant way to understand when
-        // the block was actually flushed to disk. This
-        // means changing how the blob-stores and repo are
-        // implemented.
-        // One option, is polling till we check it
-        // is written.
-        setTimeout(callback, 150)
-      }
-    })
     ws.end()
   }
 

test/block-service-test.js

@@ -193,5 +193,24 @@ module.exports = (repo) => {
         done()
       })
     })
+
+    it('stores and gets lots of blocks', (done) => {
+      const blocks = []
+      const count = 1000
+      while (blocks.length < count) {
+        blocks.push(new Block('hello-' + Math.random()))
+      }
+
+      bs.addBlocks(blocks, (err) => {
+        expect(err).to.not.exist
+
+        bs.getBlocks(blocks.map((b) => b.key), (err, res) => {
+          expect(err).to.not.exist
+          expect(Object.keys(res)).to.have.length(count)
+
+          done()
+        })
+      })
+    })
   })
 }