reimplement the the Pinned.load API

* recurse into directories on demand
* optimized for lower memory consumption
* less parallelism than before

lib/pins.js

@@ -2,6 +2,11 @@
 
 var Pins = module.exports;
 
+const Fs = require("fs");
+const Path = require("path");
+const Util = require("./common-util");
+const Plan = require("./plan");
+
 /*  Accepts a reference to an object, and...
     either a string describing which log is being processed (backwards compatibility),
     or a function which will log the error with all relevant data
@@ -22,7 +27,10 @@ var createLineHandler = Pins.createLineHandler = function (ref, errorHandler) {
     // make sure to get ref.pins as the result
     // it's a weird API but it's faster than unpinning manually
     var pins = ref.pins = {};
+    ref.index = 0;
+    ref.latest = 0;
     return function (line) {
+        ref.index++;
         if (!Boolean(line)) { return; }
 
         var l;
@@ -36,6 +44,10 @@ var createLineHandler = Pins.createLineHandler = function (ref, errorHandler) {
             return void errorHandler('PIN_LINE_NOT_FORMAT_ERROR', l);
         }
 
+        if (typeof(l[2]) === 'number') {
+            ref.latest = l[2]; // date
+        }
+
         switch (l[0]) {
             case 'RESET': {
                 pins = ref.pins = {};
@@ -72,5 +84,87 @@ Pins.calculateFromLog = function (pinFile, fileName) {
     return Object.keys(ref.pins);
 };
 
-// TODO refactor to include a streaming version for use in rpc.js as well
+/*
+    pins/
+    pins/A+/
+    pins/A+/A+hyhrQLrgYixOomZYxpuEhwfiVzKk1bBp+arH-zbgo=.ndjson
+*/
 
+Pins.list = function (done, config) {
+    const pinPath = config.pinPath || './data/pins';
+    const plan = Plan(5);
+
+    // TODO externalize this via optional handlers?
+    const stats = {
+        logs: 0,
+        dirs: 0,
+        pinned: 0,
+        lines: 0,
+    };
+
+    const errorHandler = function (label, info) {
+        console.log(label, info);
+    };
+
+    // TODO replace this with lib-readline?
+    const streamFile = function (path, cb) {
+        return void Fs.readFile(path, 'utf8', function (err, body) {
+            if (err) { return void cb(err); }
+            const ref = {};
+            const pinHandler = createLineHandler(ref, errorHandler);
+            var lines = body.split('\n');
+            stats.lines += lines.length;
+            lines.forEach(pinHandler);
+            cb(void 0, ref);
+        });
+    };
+
+    const scanDirectory = function (path, cb) {
+        Fs.readdir(path, function (err, list) {
+            if (err) {
+                return void cb(err);
+            }
+            cb(void 0, list.map(function (item) {
+                return Path.join(path, item);
+            }));
+        });
+    };
+
+    const pinned = {};
+
+    scanDirectory(pinPath, function (err, paths) {
+        if (err) { return; } // XXX
+        paths.forEach(function (path) {
+            plan.job(1, function (next) {
+                scanDirectory(path, function (nested_err, nested_paths) {
+                    if (nested_err) { return; } // XXX
+                    stats.dirs++;
+                    nested_paths.forEach(function (nested_path) {
+                        if (!/\.ndjson$/.test(nested_path)) { return; }
+                        plan.job(0, function (next)  {
+                            streamFile(nested_path, function (err, ref) {
+                                if (err) { return; } // XXX
+                                stats.logs++;
+
+                                var set = ref.pins;
+                                for (var item in set) {
+                                    if (!pinned.hasOwnProperty(item)) {
+                                        pinned[item] = true;
+                                        stats.pinned++;
+                                    }
+                                }
+                                next();
+                            });
+                        });
+                    });
+                    next();
+                });
+            });
+        });
+
+        plan.done(function () {
+            // err ?
+            done(void 0, pinned);
+        }).start();
+    });
+};