rx used

fosojs · Mar 12, 2016 · 4c870a0 · 4c870a0
1 parent aaaac04
commit 4c870a0
Show file tree

Hide file tree

Showing 2 changed files with 75 additions and 71 deletions.
diff --git a/app/plugins/bundle-service/index.js b/app/plugins/bundle-service/index.js
@@ -7,6 +7,7 @@ const registry = require('./registry')
 const chalk = require('chalk')
 const debug = require('debug')('cdn')
 const R = require('ramda')
+const Rx = require('rx')
 
 exports.register = function (plugin, opts) {
   if (!opts.storagePath) {
@@ -33,10 +34,8 @@ exports.register = function (plugin, opts) {
     if (!matchingPkg) {
       debug('No matching version found for ' +
         chalk.blue(pkgMeta.name + '@' + pkgMeta.version))
-      return Promise.reject(
-        new Error('no matching version found for ' + pkgMeta.name + '@' +
+      return new Error('no matching version found for ' + pkgMeta.name + '@' +
           pkgMeta.version)
-      )
     }
     if (matchingPkg.version !== pkgMeta.version) {
       debug(chalk.blue(pkgMeta.name + '@' + pkgMeta.version) +
@@ -49,19 +48,19 @@ exports.register = function (plugin, opts) {
         ' is overriden locally with ' +
         chalk.magenta(overrides[pkgMeta.name].path))
       const pkg = localPackage(overrides[pkgMeta.name].path)
-      return Promise.resolve({
+      return {
         pkg,
         isOverriden,
-      })
+      }
     }
     const pkg = createPackage(pkgMeta.name, matchingPkg.version, {
       registry: opts.registry,
       storagePath,
     })
-    return Promise.resolve({
+    return {
       pkg,
       isOverriden,
-    })
+    }
   }
 
   function getMatchingPkg (registryClient, pkgMeta) {
@@ -72,16 +71,16 @@ exports.register = function (plugin, opts) {
   }
 
   function fetchResources (opts) {
-    let mpkg
     const reg = registry({
       registry: opts.registry,
     })
-    return getMatchingPkg(reg, opts.pkgMeta)
-      .then(R.compose(
-        matchingPkg => getPackageLoader(opts.pkgMeta, matchingPkg, opts),
-        R.tap(value => mpkg = value))
-      )
-      .then(res => Promise.resolve(R.merge(res, {matchingPkg: mpkg})))
+    const matchingPkg$ = Rx.Observable.fromPromise(getMatchingPkg(reg, opts.pkgMeta))
+
+    return Rx.Observable.combineLatest(
+      matchingPkg$,
+      matchingPkg$.map(matchingPkg => getPackageLoader(opts.pkgMeta, matchingPkg, opts)),
+      (matchingPkg, packageLoader) => R.merge(packageLoader, {matchingPkg})
+    )
   }
 
   plugin.expose('get', function (packages, opts, cb) {
@@ -98,59 +97,60 @@ exports.register = function (plugin, opts) {
 
     const end = '.' + opts.extension
 
-    async.series(packages.map(pkgMeta => function (cb) {
-      function getMainFile (matchingPkg) {
-        const mainField = mainFields[opts.extension]
-        const mainFile = matchingPkg[mainField]
-        debug('File not specified. Loading main file: ' +
-          chalk.magenta(mainFile))
-        if (mainFile.indexOf(end) !== -1)
-          return mainFile
+    function getMainFile (matchingPkg) {
+      const mainField = mainFields[opts.extension]
+      const mainFile = matchingPkg[mainField]
+      debug('File not specified. Loading main file: ' +
+        chalk.magenta(mainFile))
+      if (mainFile.indexOf(end) !== -1)
+        return mainFile
 
-        return mainFile + end
-      }
+      return mainFile + end
+    }
 
-      function getFiles (matchingPkg) {
-        if (pkgMeta.files && pkgMeta.files.length)
-          return pkgMeta.files
+    Rx.Observable.for(packages, Rx.Observable.just)
+      .flatMapWithMaxConcurrent(1, pkgMeta => {
+        function getFiles (matchingPkg) {
+          if (pkgMeta.files && pkgMeta.files.length)
+            return pkgMeta.files
 
-        return [getMainFile(matchingPkg)]
-      }
+          return [getMainFile(matchingPkg)]
+        }
 
-      fetchResources({
-        pkgMeta,
-        registry: opts.registry,
-      }).then(function (params) {
-        const matchingPkg = params.matchingPkg
-        const isOverriden = params.isOverriden
-        const pkg = params.pkg
-        const files = getFiles(matchingPkg)
-        async.series(files.map(filePath => function (cb) {
-          pkg.readFile(filePath)
-            .then(content => cb(null, opts.transformer({
-              content,
+        return fetchResources({
+          pkgMeta,
+          registry: opts.registry,
+        })
+        .flatMap(params => {
+          return Rx.Observable.for(getFiles(params.matchingPkg), Rx.Observable.just)
+            .flatMapWithMaxConcurrent(1, filePath =>
+              Rx.Observable.fromPromise(params.pkg.readFile(filePath))
+                .map(content => ({content, filePath}))
+            )
+            .map(file => opts.transformer({
+              content: file.content,
               pkg: {
-                name: matchingPkg.name,
-                version: matchingPkg.version,
-                filePath,
+                name: params.matchingPkg.name,
+                version: params.matchingPkg.version,
+                filePath: file.filePath,
               },
-            }).content))
-            .catch(cb)
-        }), function (err, files) {
-          if (err) {
-            return cb(err)
-          }
-          cb(null, {
-            name: matchingPkg.name,
-            version: matchingPkg.version,
-            files,
-            maxAge: isOverriden ?
-              0 : plugin.plugins.fileMaxAge.getByExtension(opts.extension),
-          })
+            }))
+            .pluck('content')
+            .reduce(R.concat, [])
+            .map(files => ({
+              name: params.matchingPkg.name,
+              version: params.matchingPkg.version,
+              files,
+              maxAge: params.isOverriden ?
+                0 : plugin.plugins.fileMaxAge.getByExtension(opts.extension),
+            }))
         })
       })
-      .catch(cb)
-    }), cb)
+      .reduce(R.concat, [])
+      .subscribe(
+        result => cb(null, result),
+        cb
+      )
   })
 
   plugin.expose('getRaw', function (pkgMeta, opts, cb) {
@@ -159,21 +159,24 @@ exports.register = function (plugin, opts) {
       throw new Error('opts.registry is required')
     }
 
-    let isOverriden
-    fetchResources({
+    const fetchResources$ = fetchResources({
       pkgMeta,
       registry: opts.registry,
     })
-    .then(R.compose(
-      params => params.pkg.streamFile(pkgMeta.file),
-      R.tap(params => isOverriden = params.isOverriden)
-    ))
-    .then(stream => cb(null, {
-      stream,
-      maxAge: isOverriden ?
-        0 : plugin.plugins.fileMaxAge.getByPath(pkgMeta.file),
-    }))
-    .catch(cb)
+
+    return Rx.Observable.combineLatest(
+      fetchResources$.pluck('isOverriden'),
+      fetchResources$.flatMap(params => params.pkg.streamFile(pkgMeta.file)),
+      (isOverriden, stream) => ({
+        stream,
+        maxAge: isOverriden ?
+          0 : plugin.plugins.fileMaxAge.getByPath(pkgMeta.file),
+      })
+    )
+    .subscribe(
+      result => cb(null, result),
+      cb
+    )
   })
 }
 

diff --git a/package.json b/package.json
@@ -81,6 +81,7 @@
     "npm-registry-client": "^7.0.9",
     "plugiator": "^0.1.1",
     "ramda": "^0.19.1",
+    "rx": "^4.1.0",
     "semver": "^5.1.0",
     "stream-to-string": "^1.0.1",
     "tar-fs": "^1.10.0",