Merge pull request #54 from littlstar/updates

cleaned code, improved documentation, fixed variable naming in reduce

Author: wellsjo

lib/Request.js

@@ -33,6 +33,8 @@ class Request {
 
   /**
    * Enable modifications to the initial context.
+   *
+   * @return {Promise}
    */
 
   resolveSources() {
@@ -45,12 +47,29 @@ class Request {
     })
   }
 
+  /**
+   * Creates a progress bar for the request.
+   *
+   * @param {Batch} batch The Batch instance used for the request.
+   * @param {Number} numSources The number of files being processed.
+   */
+
+  handleProgress(batch, numSources) {
+    if (this.showProgress) {
+      const progress = new ProgressBar('[:bar] :percent', {
+        total: numSources,
+        width: 40
+      })
+      batch.on('progress', () => progress.tick())
+    }
+  }
+
   /**
    * Sets the encoding to use when getting s3 objects with
-   * <code>object.Body.toString(encoding)</code>. If not set, <code>utf8</code>
-   * is used.
+   * object.Body.toString(encoding). If not set, utf8 is used.
    *
    * @param {String} encoding The encoding.
+   * @return {Request} The instance on which this method was called.
    */
 
   encode(encoding) {
@@ -60,11 +79,12 @@ class Request {
 
   /**
    * Sets a transformation function to be used when getting objects from s3.
-   * Using <code>transform</code> takes precedence over <code>encode</code>.
+   * Using transform takes precedence over encode.
    *
    * @param {Function} transformer The function to use to transform the
    * object. The transforation function takes an s3 object as a parameter
    * and should return the file's contents as a string.
+   * @return {Request} The instance on which this method was called.
    */
 
   transform(t) {
@@ -77,6 +97,7 @@ class Request {
    * the computer can handle). Has no effect with reduce.
    *
    * @param {Integer} concurrency The concurrency level to use in the request.
+   * @return {Request} The instance on which this method was called.
    */
 
   concurrency(concurrency) {
@@ -86,7 +107,9 @@ class Request {
 
   /**
    * Limits the number of sources being operated on.
+   *
    * @param {Integer} limit
+   * @return {Request} The instance on which this method was called.
    */
 
   limit(limit) {
@@ -96,6 +119,8 @@ class Request {
 
   /**
    * Reverse the sources being operated on.
+   *
+   * @return {Request} The instance on which this method was called.
    */
 
   reverse() {
@@ -105,6 +130,8 @@ class Request {
 
   /**
    * Enables destructive actions (map, filter) to occur inplace.
+   *
+   * @return {Request} The instance on which this method was called.
    */
 
   inplace() {
@@ -119,6 +146,7 @@ class Request {
    *
    * @param {String} bucket The target bucket.
    * @param {String} prefix The target prefix (folder) where the output will go.
+   * @return {Request} The instance on which this method was called.
    */
 
   output(bucket, prefix) {
@@ -136,6 +164,9 @@ class Request {
    * @param {Function} func The function to perform over the working context.
    * @param {Boolean} [isasync=false] Set to true if `func` is async (returns a
    * Promise).
+   * @return {Promise<Object>} Resolves after processing has completed,
+   * returning an object that contains the bucket, prefix, and key of the last
+   * S3 object that was processed.
    */
 
   forEach(func, isasync) {
@@ -148,6 +179,9 @@ class Request {
    * @param {Function} func The function to perform over the working context.
    * @param {Boolean} [isAsync=false] Set to true if `func` is async (returns a
    * Promise).
+   * @return {Promise<Object>} Resolves after processing has completed,
+   * returning an object that contains the bucket, prefix, and key of the last
+   * S3 object that was processed.
    */
 
   each(func, isAsync, concurrency) {
@@ -158,22 +192,16 @@ class Request {
     return new Promise((success, fail) => {
       this.resolveSources().then((sources) => {
 
-        const progress = new ProgressBar('each [:bar] :percent', {
-          total: sources.length,
-          width: 40
-        })
-        const last = sources[sources.length - 1]
+        this.handleProgress(batch, sources.length)
+        const lastKey = sources[sources.length - 1]
 
-        // create functions array
+        // Loop over sources, apply function to each
         sources.forEach((source) => {
-
+          const bucket = source.bucket
+          const key = source.key
+          const encoding = this.opts.encoding
+          const transformer = this.opts.transformer
           batch.push((done) => {
-
-            const bucket = source.bucket
-            const key = source.key
-            const encoding = this.opts.encoding
-            const transformer = this.opts.transformer
-
             this.s3.get(bucket, key, encoding, transformer).then((body) => {
               if (isAsync) {
                 func(body, key).then(done).catch(done)
@@ -185,18 +213,14 @@ class Request {
           })
         })
 
-        if (this.showProgress) {
-          batch.on('progress', () => progress.tick())
-        }
-
+        // Resolve error or last key processed.
         batch.end((err) => {
           if (err) {
             fail(err)
           } else {
-            success(last)
+            success(lastKey)
           }
         })
-
       }).catch(fail)
     })
   }
@@ -211,6 +235,9 @@ class Request {
    * string that will replace the given s3 object.
    * @param {Boolean} [isAsync=false] If set to true, this indicates that func
    * is async and returns a promise.
+   * @return {Promise<Object>} Resolves after processing has completed,
+   * returning an object that contains the bucket, prefix, and key of the last
+   * S3 object that was processed.
    */
 
   map(func, isAsync) {
@@ -248,20 +275,16 @@ class Request {
     return new Promise((success, fail) => {
       this.resolveSources().then((sources) => {
 
-        const progress = new ProgressBar('map [:bar] :percent', {
-          total: sources.length,
-          width: 40
-        })
+        this.handleProgress(batch, sources.length)
         const lastKey = sources[sources.length - 1]
 
+        // Loop over sources, apply mapper function to each
         sources.forEach((source) => {
+          const bucket = source.bucket
+          const key = source.key
+          const encoding = this.opts.encoding
+          const transformer = this.opts.transformer
           batch.push((done) => {
-
-            const bucket = source.bucket
-            const key = source.key
-            const encoding = this.opts.encoding
-            const transformer = this.opts.transformer
-
             this.s3.get(bucket, key, encoding, transformer).then((val) => {
               if (isAsync) {
                 func(val, source.key).then((newval) => {
@@ -275,10 +298,7 @@ class Request {
           })
         })
 
-        if (this.showProgress) {
-          batch.on('progress', () => progress.tick())
-        }
-
+        // Resolve error or last key processed
         batch.end((err) => {
           if (err) {
             fail(err)
@@ -294,62 +314,60 @@ class Request {
    *
    * @param {Function} func Function to execute on each value in the array, taking
    * three arguments:
-   *   previousValue - The value previously returned in the last invocation of
-   *   func
-   *   currentValue  - The current entry being processed
-   *   key           - The key of the current object being processed
+   *   `accumulator`: The accumulated value previously returned from the last
+   *   invocation of func
+   *   `currentValue`: The current entry being processed
+   *   `key`: The key of the current object being processed
    *   func either returns the updated value, or a promise that resolves to the
    *   updated value.
-   * @param {String} value Optional.  Initial value to use as the first argument
+   * @param {String} initialValue Optional. Initial value to use as the first
+   * `previousValue` in `func`. Defaults to `null`.
    * @param {Boolean} isAsync Optional, defaults to false. If set to true, this
    * indicates that func returns a promise.
+   * @return {Promise<Object>} Resolves after processing has completed,
+   * returning the result of the reducer.
    */
 
-  reduce(func, val, isAsync) {
+  reduce(func, initialValue, isAsync) {
 
     isAsync = isAsync || this.opts.async
+    initialValue = initialValue || null
     const batch = new Batch()
-    batch.concurrency(1)
+    batch.concurrency(this.opts.concurrency)
 
     return new Promise((success, fail) => {
       this.resolveSources().then((sources) => {
 
-        const progress = new ProgressBar('reduce [:bar] :percent', {
-          total: sources.length,
-          width: 40
-        })
+        this.handleProgress(batch, sources.length)
+        let accumulator = initialValue
 
+        // Loop over sources, update `accumulator`
         sources.forEach((source) => {
+          const bucket = source.bucket
+          const key = source.key
+          const encoding = this.opts.encoding
+          const transformer = this.opts.transformer
           batch.push((done) => {
-
-            const bucket = source.bucket
-            const key = source.key
-            const encoding = this.opts.encoding
-            const transformer = this.opts.transformer
-
             this.s3.get(bucket, key, encoding, transformer).then((body) => {
               if (isAsync) {
-                func(val, body, key).then((newval) => {
-                  val = newval
+                func(accumulator, body, key).then((newval) => {
+                  accumulator = newval
                   done()
                 }).catch(done)
               } else {
-                val = func(val, body, key)
+                accumulator = func(accumulator, body, key)
                 done()
               }
             }).catch(done)
           })
         })
 
-        if (this.showProgress) {
-          batch.on('progress', () => progress.tick())
-        }
-
+        // Resolve error or reducer result `accumulator`
         batch.end((err) => {
           if (err) {
             fail(err)
           } else {
-            success(val)
+            success(accumulator)
           }
         })
       }).catch(fail)
@@ -365,6 +383,9 @@ class Request {
    * false.
    * @param {Boolean} isAsync Optional, defaults to false. If set to true, this
    * indicates that func returns a promise.
+   * @return {Promise<Object>} Resolves after processing has completed,
+   * returning an object that contains the bucket, prefix, and key of the last
+   * S3 object that was processed.
    */
 
   filter(func, isAsync) {
@@ -419,22 +440,17 @@ class Request {
     return new Promise((success, fail) => {
       this.resolveSources().then((sources) => {
 
-        const progress = new ProgressBar('filter [:bar] :percent', {
-          total: sources.length,
-          width: 40
-        })
+        this.handleProgress(batch, sources.length)
+        const lastKey = sources[sources.length - 1]
 
-        // loop over every key and run the filter function on each object. keep
+        // Loop over every key and run the filter function on each object. keep
         // track of files to keep and remove.
         sources.forEach((source) => {
-
+          const bucket = source.bucket
+          const key = source.key
+          const encoding = this.opts.encoding
+          const transformer = this.opts.transformer
           batch.push((done) => {
-
-            const bucket = source.bucket
-            const key = source.key
-            const encoding = this.opts.encoding
-            const transformer = this.opts.transformer
-
             this.s3.get(bucket, key, encoding, transformer).then((body) => {
               if (isAsync) {
                 func(body, source).then((result) => {
@@ -459,15 +475,12 @@ class Request {
           })
         })
 
-        if (this.showProgress) {
-          batch.on('progress', () => progress.tick())
-        }
-
+        // Resolve error or last source processed
         batch.end((err) => {
           if (err) {
             fail(err)
           } else {
-            success()
+            success(lastKey)
           }
         })
       }).catch(fail)