alternate sources experiment [do not merge]

.gitignore

@@ -1,4 +1,5 @@
 node_modules
+data/*
 *.log
 coverage
 .nyc_output

examples/count/index.js

@@ -9,7 +9,14 @@ tileReduce({
   bbox: [-122.05862045288086, 36.93768132842635, -121.97296142578124, 37.00378647456494],
   zoom: 15,
   map: path.join(__dirname, '/count.js'),
-  sources: [{name: 'osm', mbtiles: path.join(__dirname, '../../test/fixtures/osm.mbtiles'), raw: true}]
+  sources: [
+    {
+      name: 'osm',
+      type: 'mbtiles',
+      mbtiles: path.join(__dirname, '../../test/fixtures/osm.mbtiles'),
+      raw: true
+    }
+  ]
 })
 .on('reduce', function(num) {
   numFeatures += num;

examples/satellite-hue/index.js

@@ -0,0 +1,24 @@
+'use strict';
+
+var tileReduce = require('../../src');
+var fs = require('fs');
+var path = require('path');
+
+var numFeatures = 0;
+
+if (!fs.existsSync(path.join(__dirname, '../../data'))) fs.mkdirSync(path.join(__dirname, '../../data'));
+
+tileReduce({
+  bbox: [-122.02, 36.98, -122.0, 37.0],
+  zoom: 15,
+  map: path.join(__dirname, '/map.js'),
+  sources: [
+    {
+      type: 'remote',
+      name: 'satellite',
+      url: 'https://b.tiles.mapbox.com/v4/mapbox.satellite/{z}/{x}/{y}.png?access_token=' + process.env.MapboxAccessToken,
+      raw: false
+    }
+  ]
+});
+

examples/satellite-hue/map.js

@@ -0,0 +1,11 @@
+'use strict';
+
+var path = require('path');
+
+module.exports = function(data, tile, writeData, done) {
+  data.satellite.hue(180, function (err, img) {
+    img.writeFile(path.join(__dirname, '../../data', tile[0] + '-' + tile[1] + '-' + tile[2] + '.jpg'), 'jpg', function () {
+      done();
+    });
+  });
+};

package.json

@@ -32,6 +32,7 @@
     "through2": "^2.0.0",
     "tile-cover": "^3.0.1",
     "tilebelt": "^1.0.1",
+    "tilelive": "^5.12.2",
     "turf-bbox-polygon": "^1.0.1",
     "vector-tile": "^1.1.3"
   },

src/adapters/mbtiles.js

@@ -0,0 +1,42 @@
+'use strict';
+
+var zlib = require('zlib');
+var MBTiles = require('mbtiles');
+var parseVT = require('./vt');
+
+module.exports = function (options, callback) {
+  var self = this;
+
+  this.options = options;
+  this.db = new MBTiles(options.mbtiles, dbReady);
+  function dbReady(err, db) {
+    if (err) callback(err);
+    else self.db.getInfo(infoReady);
+  }
+
+  function infoReady(err, info) {
+    if (err) {
+      callback(err);
+    } else if (info.format === 'pbf') {
+      callback(null, self);
+    } else {
+      callback(new Error('Unsupported MBTiles format: ' + info.format));
+    }
+  }
+};
+
+module.exports.prototype.getTile = function (z, x, y, callback) {
+  var self = this;
+  this.db.getTile(z, x, y, tileFetched);
+
+  function tileFetched(err, data) {
+    if (!err) zlib.unzip(data, tileUnzipped);
+    else if (err.message === 'Tile does not exist') callback();
+    else callback(err);
+  }
+
+  function tileUnzipped(err, data) {
+    if (err) callback(err);
+    callback(null, parseVT(data, [x, y, z], self.options));
+  }
+};

src/adapters/remote.js

@@ -0,0 +1,43 @@
+'use strict';
+
+var request = require('request');
+var parseVT = require('./vt');
+var lwip = require('lwip');
+var rateLimit = require('function-rate-limit');
+
+module.exports = function (options, callback) {
+  this.options = options;
+  if (options.maxrate) this.getTile = rateLimit(options.source.maxrate, 1000, getTile);
+  callback(null, this);
+};
+
+module.exports.prototype.getTile = function (z, x, y, done) {
+  var self = this;
+  var url = this.options.url
+    .replace('{x}', x)
+    .replace('{y}', y)
+    .replace('{z}', z);
+
+  request({url: url, gzip: true, encoding: null}, function(err, res, body) {
+    if (err) return done(err);
+    else if (res.statusCode === 200) {
+      var ctype = res.headers['content-type'].split('/')[1];
+
+      // if content type is protobuf, read as a vector tile
+      if (ctype === 'x-protobuf') {
+        return done(null, parseVT(body, [x, y, z], self.options));
+      } else {
+        if (self.options.raw) return done(null, body);
+
+        // otherwise, assume it's an image tile, open it with lwip
+        lwip.open(body, ctype, function (err, img) {
+          if (err) return done(err);
+          done(null, img);
+        });
+      }
+    }
+    else if (res.statusCode === 401) return done();
+    else if (res.statusCode === 404) return done();
+    else return done(new Error('Server responded with status code ' + res.statusCode));
+  });
+};

src/index.js

@@ -30,6 +30,10 @@ function tileReduce(options) {
   var pauseLimit = options.batch || 5000;
   var start = Date.now();
   var timer;
+  var adapters = {
+    'mbtiles': path.join(__dirname, 'adapters/mbtiles'),
+    'remote': path.join(__dirname, 'adapters/remote')
+  };
 
   // Validate syntax in the map script to fail faster
   try {
@@ -57,7 +61,7 @@ function tileReduce(options) {
   var mapOptions = options.mapOptions || {};
 
   for (var i = 0; i < maxWorkers; i++) {
-    var worker = fork(path.join(__dirname, 'worker.js'), [options.map, JSON.stringify(options.sources), JSON.stringify(mapOptions)], {silent: true});
+    var worker = fork(path.join(__dirname, 'worker.js'), [options.map, JSON.stringify(options.sources), JSON.stringify(adapters), JSON.stringify(mapOptions)], {silent: true});
     worker.stdout.pipe(binarysplit('\x1e')).pipe(output);
     worker.stderr.pipe(process.stderr);
     worker.on('message', handleMessage);

src/worker.js

@@ -1,39 +1,45 @@
 'use strict';
 
+var path = require('path');
 var queue = require('queue-async');
 var q = queue();
 var sources = [];
 var tilesQueue = queue(1);
 var isOldNode = process.versions.node.split('.')[0] < 4;
 
-global.mapOptions = JSON.parse(process.argv[4]);
+var adapters = JSON.parse(process.argv[4]);
+for (var key in adapters) {
+  // eslint-disable-line global-require
+  adapters[key] = require(adapters[key]);
+}
+
+global.mapOptions = JSON.parse(process.argv[5]);
 var map = require(process.argv[2]);
 
 JSON.parse(process.argv[3]).forEach(function(source) {
   q.defer(loadSource, source);
 });
 
-function loadSource(source, done) {
-  var loaded = {name: source.name};
-  sources.push(loaded);
 
+function loadSource(source, done) {
   /*eslint global-require: 0 */
-  if (source.mbtiles) require('./mbtiles')(source, done);
-  else if (source.url) require('./remote')(source, done);
-  else throw new Error('Unknown source type');
+  if (!adapters[source.type]) throw new Error('Unknown source type ' + source.type);
+
+  var adapter = new adapters[source.type](source, done);
+  adapter.name = source.name;
 }
 
 q.awaitAll(function(err, results) {
   if (err) throw err;
-  for (var i = 0; i < results.length; i++) sources[i].getTile = results[i];
+  sources = results;
   process.send({ready: true});
 });
 
 function processTile(tile, callback) {
   var q = queue();
 
   for (var i = 0; i < sources.length; i++) {
-    q.defer(sources[i].getTile, tile);
+    q.defer(sources[i].getTile.bind(sources[i]), tile[2], tile[0], tile[1]);
   }
 
   q.awaitAll(gotData);
@@ -50,7 +56,6 @@ function processTile(tile, callback) {
         return;
       }
     }
-
     var writeQueue = queue(1);
 
     function write(data) {