feat: Merge pull request #163 from pelias/async-file-loading

Use Async methods for file loading

package.json

@@ -37,6 +37,7 @@
     "pelias-dbclient": "2.0.0",
     "pelias-logger": "0.1.0",
     "pelias-model": "4.3.0",
+    "pelias-parallel-stream": "0.0.2",
     "through2": "^2.0.0",
     "through2-filter": "^2.0.0",
     "through2-map": "^2.0.0",

src/components/loadJSON.js

@@ -1,16 +1,25 @@
-var map = require('through2-map');
 var fs = require('fs');
+var parallelStream = require('pelias-parallel-stream');
+
+var maxInFlight = 10;
 
 module.exports.create = function create_json_parse_stream(dataDirectory) {
-  return map.obj(function(record) {
-    try {
-      return JSON.parse(fs.readFileSync(dataDirectory + record.path));
-    } catch (err) {
-      console.error('exception on %s:', record.path, err);
-      console.error('Inability to parse JSON usually means that WOF has been cloned ' +
-                    'without using git-lfs, please see instructions here: ' +
-                    'https://github.com/whosonfirst/whosonfirst-data#git-and-large-files');
-      return {};
-    }
+  return parallelStream(maxInFlight, function(record, enc, next) {
+    fs.readFile(dataDirectory + record.path, function(err, data) {
+      if (err) {
+        next(err);
+      } else {
+        try {
+          var object = JSON.parse(data);
+          next(null, object);
+        } catch (parse_err) {
+          console.error('exception on %s:', record.path, parse_err);
+          console.error('Inability to parse JSON usually means that WOF has been cloned ' +
+                        'without using git-lfs, please see instructions here: ' +
+                        'https://github.com/whosonfirst/whosonfirst-data#git-and-large-files');
+          next(null, {});
+        }
+      }
+    });
   });
 };

src/readStream.js

@@ -6,7 +6,6 @@ var through2 = require('through2');
 var logger = require( 'pelias-logger' ).get( 'whosonfirst' );
 
 var isValidId = require('./components/isValidId');
-var fileIsReadable = require('./components/fileIsReadable');
 var loadJSON = require('./components/loadJSON');
 var recordHasIdAndProperties = require('./components/recordHasIdAndProperties');
 var isActiveRecord = require('./components/isActiveRecord');
@@ -59,7 +58,6 @@ function createReadStream(directory, types, wofAdminRecords) {
 
   return createMetaRecordStream(metaFilePaths, types)
   .pipe(isValidId.create())
-  .pipe(fileIsReadable.create(directory + 'data/'))
   .pipe(loadJSON.create(directory + 'data/'))
   .pipe(recordHasIdAndProperties.create())
   .pipe(isActiveRecord.create())

test/test.js

@@ -1,5 +1,4 @@
 require ('./components/extractFieldsTest.js');
-require ('./components/fileIsReadableTest.js');
 require ('./components/isActiveRecordTest.js');
 require ('./components/isValidIdTest.js');
 require ('./components/loadJSONTest.js');