-
Notifications
You must be signed in to change notification settings - Fork 20
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #51 from osmlab/tests
tests / async refactor
- Loading branch information
Showing
14 changed files
with
1,720 additions
and
179 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,31 +1,50 @@ | ||
| var fs = require('fs'), | ||
| levelup = require('levelup'), | ||
| leveldown = require('leveldown'), | ||
| key = require('./lib/key.js'); | ||
| queue = require('queue-async'), | ||
| key = require('./lib/key.js'), | ||
| rimraf = require('rimraf'); | ||
|
|
||
| var dbs = fs.readdirSync('./ldb/').filter(function(item) { | ||
| return item.indexOf('.ldb') > -1; | ||
| }); | ||
| function fixed(callback) { | ||
| var dbs = fs.readdirSync('./ldb/').filter(function(item) { | ||
| return item.indexOf('.ldb') > -1; | ||
| }); | ||
|
|
||
| if (!fs.existsSync('./fixed')) { | ||
| fs.mkdirSync('./fixed'); | ||
| } | ||
| if (!fs.existsSync('./fixed')) { | ||
| fs.mkdirSync('./fixed'); | ||
| } | ||
|
|
||
| dbs.forEach(function(ldb, idx) { | ||
| levelup('./ldb/' + ldb, { | ||
| createIfMissing: false, | ||
| max_open_files: 500 | ||
| }, function(err, db) { | ||
| if (err) return console.log(err); | ||
| var file = fs.createWriteStream('./fixed/' + ldb.split('.ldb')[0]); | ||
| file.once('open', function() { | ||
| db.createReadStream({lt: '0001'}) | ||
| .on('data', function(data) { | ||
| file.write(key.decompose(data.key).hash + '\n'); | ||
| }) | ||
| .on('end', function(data) { | ||
| file.end(); | ||
| var q = queue(); | ||
|
|
||
| dbs.forEach(function(ldb, idx) { | ||
| q.defer(function(qcallback) { | ||
| levelup('./ldb/' + ldb, { | ||
| createIfMissing: false, | ||
| max_open_files: 500 | ||
| }, function(err, db) { | ||
| if (err) return console.log(err); | ||
| var filePath = './fixed/' + ldb.split('.ldb')[0]; | ||
| var file = fs.createWriteStream(filePath); | ||
| file.once('open', function() { | ||
| db.createReadStream({lt: '0001'}) | ||
| .on('data', function(data) { | ||
| file.write(key.decompose(data.key).hash + '\n'); | ||
| }) | ||
| .on('end', function(data) { | ||
| file.end(); | ||
| db.close(function(err) { | ||
| qcallback(err); | ||
| }); | ||
| }); | ||
| }); | ||
| }); | ||
| }); | ||
| }); | ||
| }); | ||
|
|
||
| q.awaitAll(function(err, results) { | ||
| if (callback) callback(err); | ||
| }); | ||
| } | ||
|
|
||
| module.exports = fixed; | ||
| if (require.main === module) { fixed(); } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,78 +1,121 @@ | ||
| var fs = require('fs'), | ||
| path = require('path'), | ||
| readline = require('readline'), | ||
| csv = require('csv-parser'), | ||
| levelup = require('levelup'), | ||
| key = require('./lib/key.js'); | ||
| key = require('./lib/key.js'), | ||
| queue = require('queue-async'), | ||
| rimraf = require('rimraf'); | ||
|
|
||
| if (process.stdin.isTTY) { | ||
| var verbose = false; | ||
| if (require.main === module) { | ||
| verbose = true; | ||
| if (process.argv[2] === undefined) { | ||
| return console.log('file argument required \n`node import-csv.js [source csv]`'); | ||
| } | ||
| loadTask(process.argv[2]); | ||
| } else { | ||
| process.stdin.on('readable', function() { | ||
| var buf = process.stdin.read(); | ||
| if (buf === null) return; | ||
| buf.toString().split('\n').forEach(function(file) { | ||
| if (file.length) loadTask(file, process.argv[2]); | ||
| }); | ||
| }); | ||
| loadTask(process.argv[2], function() {}); | ||
| } | ||
|
|
||
| function loadTask(fileLoc) { | ||
| var task = path.basename(fileLoc).split('.')[0], | ||
| db = levelup('./ldb/' + task + '.ldb'), | ||
| fixed_list = [], | ||
| count = 0; | ||
| module.exports = loadTask; | ||
|
|
||
| var tracking = levelup('./ldb/' + task + '-tracking.ldb'); | ||
| tracking.close(); | ||
| function loadTask(fileLoc, callback) { | ||
| var task = path.basename(fileLoc).split('.')[0], | ||
| topq = queue(); | ||
|
|
||
| if (fs.existsSync('./fixed') && fs.readdirSync('./fixed').indexOf(task) > -1) { | ||
| var rl = readline.createInterface({ | ||
| input: fs.createReadStream('./fixed/' + task), | ||
| output: new require('stream') | ||
| // ensure that we create & close the tracking database before loading anything | ||
| topq.defer(function(tqcallback) { | ||
| levelup('./ldb/' + task + '-tracking.ldb', {}, function(err, trackingdb) { | ||
| if (err) return tqcallback(err); | ||
| trackingdb.close(function(err) { | ||
| tqcallback(err); | ||
| }); | ||
| }); | ||
| }); | ||
|
|
||
| rl.on('line', function(line) { | ||
| fixed_list.push(line); | ||
| }); | ||
| topq.awaitAll(function(err, results) { | ||
| // perform some tests to see if there's a 'fixed' file for this task | ||
| var fixedq = queue(); | ||
| fixedq | ||
| .defer(function(fqcallback) { | ||
| fs.readdir('./fixed', function(err, files) { | ||
| if (err) return fqcallback(err); | ||
| fqcallback(err, (files.indexOf(task) > -1)); | ||
| }); | ||
| }) | ||
| .defer(function(fqcallback) { | ||
| fs.stat('./fixed/' + task, function(err, info) { | ||
| if (err) return fqcallback(err); | ||
| fqcallback(err, (info.size > 0)); | ||
| }); | ||
| }); | ||
|
|
||
| rl.on('end', function() { | ||
| doImport(fileLoc); | ||
| // once the tests come back, check if they're all positive. if so, load fixes & tasks. | ||
| fixedq.awaitAll(function(err, results) { | ||
| if((!err) && results.every(function(x) { return x; })) { | ||
| fs.readFile('./fixed/' + task, function(err, data) { | ||
| var fixed_list = data.toString().split("\n"); | ||
| _doImport(fileLoc, task, fixed_list, callback); | ||
| }); | ||
| } else { | ||
| _doImport(fileLoc, task, [], callback); | ||
| } | ||
| }); | ||
| } else { | ||
| doImport(fileLoc); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| function doImport(fileLoc) { | ||
| console.log('importing ' + task); | ||
| fs.createReadStream(fileLoc) | ||
| .pipe(csv()) | ||
| .on('data', function(data) { | ||
| var object_hash = key.hashObject(data); | ||
| if (fixed_list.indexOf(object_hash) === -1) { | ||
| // item is not fixed | ||
| var object_id = key.compose(1, object_hash); | ||
| db.put(object_id, JSON.stringify(data), function (err) { | ||
| if (err) console.log('-- error --', err); | ||
| }); | ||
| count++; | ||
| } | ||
| }) | ||
| .on('end', function() { | ||
| setTimeout(function() { | ||
| // insert a dummy object in unfixed keyspace if nothing has been inserted | ||
| // prevents blocking on levelup readstream creation against an empty db | ||
| if (count === 0) { | ||
| function _doImport(fileLoc, task, fixed_list, callback) { | ||
| if (verbose) console.log('importing task from ' + fileLoc); | ||
| var count = 0; | ||
|
|
||
| levelup('./ldb/' + task + '.ldb', function(err, db) { | ||
| if (err) throw callback(err); | ||
| var q = queue(); | ||
|
|
||
| q.defer(function(qcallback) { | ||
| fs.createReadStream(fileLoc) | ||
| .pipe(csv()) | ||
| .on('data', function(data) { | ||
| var object_hash = key.hashObject(data); | ||
| if (fixed_list.indexOf(object_hash) === -1) { | ||
| // item is not fixed | ||
| var object_id = key.compose(1, object_hash); | ||
| count++; | ||
| q.defer(function(qsubcallback) { | ||
| db.put(object_id, JSON.stringify(data), function(err) { | ||
| qsubcallback(err); | ||
| }); | ||
| }); | ||
| } | ||
| }) | ||
| .on('end', function(err) { | ||
| if(count === 0) { | ||
| var keyval = key.compose(1, 'random'); | ||
| db.put(keyval, JSON.stringify({ignore: true})); | ||
| q.defer(function(qsubcallback) { | ||
| db.put(keyval, JSON.stringify({ignore: true}), function(err) { | ||
| qsubcallback(err); | ||
| }); | ||
| }); | ||
| } | ||
| db.close(); | ||
|
|
||
| console.log('done with ' + task + '. ' + count + ' items imported'); | ||
| }, 5000); | ||
| qcallback(err); | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| q.awaitAll(function(err, results) { | ||
| if (verbose) { console.log('done with ' + task + '. ' + count + ' items imported'); } | ||
| db.close(function(err) { | ||
| if (callback) callback(err); | ||
| }); | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| /* | ||
| function deleteTask(task, callback){ | ||
| var q = queue(); | ||
| if (verbose) { console.log('deleting task ' + task); } | ||
| q.defer(rimraf, './ldb/' + task + '.ldb') | ||
| .defer(rimraf, './ldb/' + task + '-tracking.ldb') | ||
| .awaitAll(function(err, results) { | ||
| callback(err, results); | ||
| }); | ||
| } | ||
| */ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| * |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| var levelup = require('levelup'), | ||
| key = require('./key.js'), | ||
| queue = require('queue-async'); | ||
|
|
||
| function countTasksBySkipval(callback) { | ||
| levelup('./ldb/test.ldb', function(err, db) { | ||
| if (err) return callback(err); | ||
| var counts = {}; | ||
| db.createReadStream() | ||
| .on('data', function(data) { | ||
| var skipval = key.decompose(data.key).skipval; | ||
| if (!counts[skipval]) counts[skipval] = 0; | ||
| counts[skipval]++; | ||
| }) | ||
| .on('end', function() { | ||
| db.close(function(err) { | ||
| callback(err, counts); | ||
| }); | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| function markTasksAsDone(tasks, callback) { | ||
| levelup('./ldb/test.ldb', function(err, db) { | ||
| if (err) throw callback(err); | ||
| var q = queue(); | ||
| tasks.forEach(function(task) { | ||
| q.defer(function(task, deferCallback) { | ||
| db.get(task, function(err, val) { | ||
| var k = key.decompose(task); | ||
| db.put(key.compose(0, k.hash), val, function() { | ||
| db.del(task, deferCallback); | ||
| }); | ||
| }); | ||
| }, task); | ||
| }); | ||
| q.awaitAll(function() { | ||
| db.close(function(err) { | ||
| callback(err); | ||
| }); | ||
|
|
||
| }); | ||
| }); | ||
| } | ||
|
|
||
| function blankslate(clearFixed, callback){ | ||
| var q = queue(); | ||
| if (clearFixed) { | ||
| q.defer(rimraf, './fixed/test'); | ||
| q.defer(rimraf, './fixed/test-tracking'); | ||
| } | ||
| q.defer(rimraf, './ldb/test.ldb'); | ||
| q.defer(rimraf, './ldb/test-tracking.ldb'); | ||
| q.awaitAll(function(err, results) { | ||
| callback(err); | ||
| }); | ||
| } | ||
|
|
||
| module.exports = { | ||
| countTasksBySkipval: countTasksBySkipval, | ||
| markTasksAsDone: markTasksAsDone, | ||
| blankslate: blankslate | ||
| }; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.