Merge pull request #43 from mapbox/zxystream

Add readable ZXYStream and createZXYStream method.

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.7.0
+
+- Adds `createZXYStream` method for a readstream of z/x/y tile coordinates.
+
 # 0.6.0
 
 - Update dep to ~sqlite@3.0.0.

lib/mbtiles.js

@@ -8,6 +8,7 @@ var Buffer = require('buffer').Buffer;
 var sm = new (require('sphericalmercator'));
 var sqlite3 = require('sqlite3');
 var tiletype = require('tiletype');
+var ZXYStream = require('./zxystream');
 
 function noop(err) {
     if (err) throw err;
@@ -740,3 +741,8 @@ MBTiles.prototype.geocoderCentroid = function(id, zxy, callback) {
         ], mid[0]));
     });
 };
+
+MBTiles.prototype.createZXYStream = function(options) {
+    return new ZXYStream(this, options);
+};
+

lib/zxystream.js

@@ -0,0 +1,46 @@
+var stream = require('stream');
+var util = require('util');
+
+module.exports = ZXYStream;
+util.inherits(ZXYStream, stream.Readable);
+
+// Readable stream of line-delimited z/x/y coordinates
+// contained within the MBTiles `tiles` table/view.
+//
+// The `batch` option exists to allow tests to check that
+// multiple calls to `_read` are handled properly. IRL the
+// default offset of 10000 should be reasonably efficient
+// and not worth messing with.
+function ZXYStream(source, options) {
+    if (!source) throw new TypeError('MBTiles source required');
+
+    options = options || {};
+
+    if (options.batch !== undefined && typeof options.batch !== 'number')
+        throw new TypeError('options.batch must be a positive integer');
+
+    this.source = source;
+    this.batch = options.batch || 10000;
+    this.offset = 0;
+
+    stream.Readable.call(this);
+}
+
+ZXYStream.prototype._read = function() {
+    var stream = this;
+    this.source._db.all('SELECT zoom_level AS z, tile_column AS x, tile_row AS y FROM tiles LIMIT ' + this.batch + ' OFFSET ' + this.offset, function(err, rows) {
+        if (err) return stream.emit('error', err);
+        if (!rows.length) return stream.push(null);
+        stream.offset += stream.batch;
+        var chunk = '';
+        for (var i = 0; i < rows.length; i++) chunk += toLine(rows[i]);
+        stream.push(chunk);
+    });
+};
+
+function toLine(row) {
+    // Flip Y coordinate because MBTiles files are TMS.
+    var y = row.y = (1 << row.z) - 1 - row.y;
+    return row.z + '/' + row.x + '/' + y + '\n';
+}
+

test/zxystream.js

@@ -0,0 +1,75 @@
+var tape = require('tape');
+var MBTiles = require('../lib/mbtiles.js');
+var source;
+
+tape('zxystream setup', function(assert) {
+    new MBTiles(__dirname + '/fixtures/plain_2.mbtiles', function(err, s) {
+        assert.ifError(err);
+        source = s;
+        assert.end();
+    });
+});
+
+tape('zxystream default batch', function(assert) {
+    var stream = source.createZXYStream();
+    var output = '';
+    var called = 0;
+
+    assert.deepEqual(stream.source, source, 'sets stream.source');
+    assert.deepEqual(stream.batch, 10000, 'sets stream.batch = 10000');
+    assert.deepEqual(stream.offset, 0, 'sets stream.offset = 0');
+
+    stream.on('data', function(lines) {
+        output += lines;
+        called++;
+    });
+    stream.on('end', function() {
+        var queue = output.toString().split('\n');
+        assert.equal(queue.length, 270);
+        assert.equal(called, 1, 'emitted data x1 times');
+        checkTile(queue);
+        function checkTile(queue) {
+            if (!queue.length) return assert.end();
+            var zxy = queue.shift();
+            if (!zxy) return checkTile(queue);
+            zxy = zxy.split('/');
+            source.getTile(zxy[0], zxy[1], zxy[2], function(err, buffer, headers) {
+                assert.equal(!err && (buffer instanceof Buffer), true, zxy.join('/') + ' exists');
+                checkTile(queue);
+            });
+        }
+    });
+});
+
+
+tape('zxystream batch = 10', function(assert) {
+    var stream = source.createZXYStream({batch:10});
+    var output = '';
+    var called = 0;
+
+    assert.deepEqual(stream.source, source, 'sets stream.source');
+    assert.deepEqual(stream.batch, 10, 'sets stream.batch = 10');
+    assert.deepEqual(stream.offset, 0, 'sets stream.offset = 0');
+
+    stream.on('data', function(lines) {
+        output += lines;
+        called++;
+    });
+    stream.on('end', function() {
+        var queue = output.toString().split('\n');
+        assert.equal(queue.length, 270);
+        assert.equal(called, 27, 'emitted data x27 times');
+        checkTile(queue);
+        function checkTile(queue) {
+            if (!queue.length) return assert.end();
+            var zxy = queue.shift();
+            if (!zxy) return checkTile(queue);
+            zxy = zxy.split('/');
+            source.getTile(zxy[0], zxy[1], zxy[2], function(err, buffer, headers) {
+                assert.equal(!err && (buffer instanceof Buffer), true, zxy.join('/') + ' exists');
+                checkTile(queue);
+            });
+        }
+    });
+});
+