feat: track progress events

Author: dryajov

examples/upload-file-via-browser/package.json

@@ -13,7 +13,7 @@
   "devDependencies": {
     "babel-core": "^5.4.7",
     "babel-loader": "^5.1.2",
-    "ipfs-api": "^12.1.7",
+    "ipfs-api": "../../",
     "json-loader": "^0.5.4",
     "react": "^15.4.2",
     "react-dom": "^15.4.2",

examples/upload-file-via-browser/src/App.js

@@ -29,7 +29,7 @@ class App extends React.Component {
   saveToIpfs (reader) {
     let ipfsId
     const buffer = Buffer.from(reader.result)
-    this.ipfsApi.add(buffer)
+    this.ipfsApi.add(buffer, { progress: (prog) => console.log(`received: ${prog}`) })
       .then((response) => {
         console.log(response)
         ipfsId = response[0].hash

src/files/add.js

@@ -3,6 +3,7 @@
 const isStream = require('is-stream')
 const promisify = require('promisify-es6')
 const DAGNodeStream = require('../utils/dagnode-stream')
+const ProgressStream = require('../utils/progress-stream')
 
 module.exports = (send) => {
   return promisify((files, opts, callback) => {
@@ -14,8 +15,8 @@ module.exports = (send) => {
     opts = opts || {}
 
     const ok = Buffer.isBuffer(files) ||
-               isStream.readable(files) ||
-               Array.isArray(files)
+      isStream.readable(files) ||
+      Array.isArray(files)
 
     if (!ok) {
       return callback(new Error('"files" must be a buffer, readable stream, or array of objects'))
@@ -41,10 +42,12 @@ module.exports = (send) => {
       qs.hash = opts.hashAlg
     }
 
-    const request = { path: 'add', files: files, qs: opts, progress: opts.progress }
+    const request = { path: 'add', files: files, qs: qs, progress: opts.progress }
 
     // Transform the response stream to DAGNode values
-    const transform = (res, callback) => DAGNodeStream.streamToValue(send, res, callback)
+    const transform = (res, callback) => DAGNodeStream.streamToValue(send,
+      ProgressStream.fromStream(opts.progress, res),
+      callback)
     send.andTransform(request, transform, callback)
   })
 }

src/utils/progress-stream.js

@@ -0,0 +1,50 @@
+'use strict'
+
+const Transform = require('readable-stream').Transform
+
+/*
+  A transform stream to track progress events on file upload
+
+  When the progress flag is passed to the HTTP api, the stream
+  emits progress events like such:
+
+  {
+    Name  string
+    Hash  string `json:",omitempty"`
+    Bytes int64  `json:",omitempty"`
+    Size  string `json:",omitempty"`
+  }
+
+  This class will take care of detecting such
+  events and calling the associated track method
+  with the bytes sent so far as parameter. It will
+  also skip them from the stream, emitting only
+  when the final object has been uploaded and we
+  got a hash.
+*/
+class ProgressStream extends Transform {
+  constructor (opts) {
+    opts = Object.assign(opts || {}, { objectMode: true })
+    super(opts)
+    this._track = opts.track || (() => {})
+    this._res = []
+  }
+
+  static fromStream (track, stream) {
+    const prog = new ProgressStream({ track })
+    return stream.pipe(prog)
+  }
+
+  _transform (chunk, encoding, callback) {
+    if (chunk &&
+      typeof chunk.Bytes !== 'undefined' &&
+      typeof chunk.Hash === 'undefined') {
+      this._track(chunk.Bytes)
+      return callback()
+    }
+
+    callback(null, chunk)
+  }
+}
+
+module.exports = ProgressStream

src/utils/request-api.js

@@ -10,7 +10,6 @@ const getFilesStream = require('./get-files-stream')
 const streamToValue = require('./stream-to-value')
 const streamToJsonValue = require('./stream-to-json-value')
 const request = require('./request')
-const Transform = require('readable-stream').Transform
 
 // -- Internal
 
@@ -82,6 +81,9 @@ function requestAPI (config, options, callback) {
   if (options.files && !Array.isArray(options.files)) {
     options.files = [options.files]
   }
+  if (options.progress) {
+    options.qs.progress = true
+  }
 
   if (options.qs.r) {
     options.qs.recursive = options.qs.r
@@ -161,17 +163,7 @@ function requestAPI (config, options, callback) {
   })
 
   if (options.files) {
-    if (options.progress && typeof options.progress === 'function') {
-      const progressStream = new Transform({
-        transform: (chunk, encoding, cb) => {
-          options.progress(chunk.byteLength)
-          cb(null, chunk)
-        }
-      })
-      stream.pipe(progressStream).pipe(req)
-    } else {
-      stream.pipe(req)
-    }
+    stream.pipe(req)
   } else {
     req.end()
   }

test/files.spec.js

@@ -107,10 +107,12 @@ describe('.files (the MFS API part)', function () {
     })
 
     it.only('files.add with progress options', (done) => {
-      ipfs.files.add(testfile, {progress: false}, (err, res) => {
+      let progress = 0
+      ipfs.files.add(testfile, { progress: (p) => { progress = p } }, (err, res) => {
         expect(err).to.not.exist()
 
         expect(res).to.have.length(1)
+        expect(progress).to.be.greaterThan(0)
         done()
       })
     })
@@ -158,7 +160,7 @@ describe('.files (the MFS API part)', function () {
 
     it('files.write', (done) => {
       ipfs.files
-        .write('/test-folder/test-file-2.txt', Buffer.from('hello world'), {create: true}, (err) => {
+        .write('/test-folder/test-file-2.txt', Buffer.from('hello world'), { create: true }, (err) => {
           expect(err).to.not.exist()
 
           ipfs.files.read('/test-folder/test-file-2.txt', (err, stream) => {
@@ -254,7 +256,7 @@ describe('.files (the MFS API part)', function () {
     })
 
     it('files.rm', (done) => {
-      ipfs.files.rm('/test-folder', {recursive: true}, done)
+      ipfs.files.rm('/test-folder', { recursive: true }, done)
     })
   })
 
@@ -330,7 +332,7 @@ describe('.files (the MFS API part)', function () {
 
     it('files.write', (done) => {
       ipfs.files
-        .write('/test-folder/test-file-2.txt', Buffer.from('hello world'), {create: true})
+        .write('/test-folder/test-file-2.txt', Buffer.from('hello world'), { create: true })
         .then(() => {
           return ipfs.files.read('/test-folder/test-file-2.txt')
         })
@@ -396,7 +398,9 @@ describe('.files (the MFS API part)', function () {
     })
 
     it('files.read', (done) => {
-      if (!isNode) { return done() }
+      if (!isNode) {
+        return done()
+      }
 
       ipfs.files.read('/test-folder/test-file')
         .then((stream) => {