Parallel copy

lib/stream-deserialize.js

@@ -9,30 +9,26 @@ var StringDecoder = require('string_decoder').StringDecoder;
 module.exports = Deserialize;
 util.inherits(Deserialize, stream.Transform);
 
-function Deserialize() {
+function Deserialize(options) {
     stream.Transform.call(this);
+
+    options = options || {};
+
     this._writableState.objectMode = false;
     this._readableState.objectMode = true;
     this._buffer = '';
     this._decoder = new StringDecoder('utf8');
+    this._job = options.job || false;
 }
 
 Deserialize.prototype._transform = function(chunk, enc, callback) {
     this._buffer += this._decoder.write(chunk);
     var lines = this._buffer.split(/\r?\n/);
     this._buffer = lines.pop();
 
-    var obj, line;
     for (var i = 0; i < lines.length; i++) {
-        line = lines[i];
-
-        if (line.toString() === serialHeader) continue;
-
-        try { obj = deserialize(line); }
-        catch (err) { return callback(err); }
-        if (obj instanceof Info) this.emit('info', obj);
-        if (obj instanceof Tile) this.emit('tile', obj)
-        this.push(obj);
+        try { this.deserialize(lines[i]); }
+        catch(err) { return callback(err); }
     }
 
     callback();
@@ -41,14 +37,24 @@ Deserialize.prototype._transform = function(chunk, enc, callback) {
 Deserialize.prototype._flush = function(callback) {
     var leftover = this._buffer.trim();
     if (leftover) {
-        var obj;
+        try { this.deserialize(leftover); }
+        catch(err) { return callback(err); }
+    }
+    callback();
+};
+
+Deserialize.prototype.deserialize = function(serializedObj) {
+    if (serializedObj.toString() === serialHeader) return;
 
-        try { obj = deserialize(leftover); }
-        catch (err) { return callback(err); }
-        if (obj instanceof Info) this.emit('info', obj);
-        if (obj instanceof Tile) this.emit('tile', obj)
-        this.push(obj);
+    if (this._job) {
+        var x = deserialize(serializedObj, 'x');
+        if (x % this._job.total !== this._job.num - 1)
+            return;
     }
 
-    callback();
+    var obj = deserialize(serializedObj);
+
+    if (obj instanceof Info) this.emit('info', obj);
+    if (obj instanceof Tile) this.emit('tile', obj);
+    this.push(obj);
 };

lib/stream-list.js

@@ -17,6 +17,7 @@ function List(source, options) {
     this.bufferzxy = [];
     this.stats = new Stats();
     this.length = 0;
+    this.job = options.job || false;
 
     // Determine when the writable stream is finished so the
     // readable stream can stop waiting.
@@ -59,23 +60,30 @@ List.prototype._read = function(size) {
         }
         var zxy = stream.bufferzxy.shift();
         stream.stats.ops++;
+
+        if (stream.job && zxy.x % stream.job.total !== stream.job.num - 1)
+            return skip();
+
         stream.source.getTile(zxy.z, zxy.x, zxy.y, function(err, buffer) {
             if (err && !(/does not exist$/).test(err.message)) {
                 stream.emit('error', err);
             } else if (err || isEmpty(buffer)) {
-                stream.stats.skipped++;
-                stream.stats.done++;
-                // Update length
-                stream.length--;
-                stream.emit('length', stream.length);
-                get(push);
+                skip();
             } else {
                 stream.stats.done++;
                 push(new Tile(zxy.z, zxy.x, zxy.y, buffer));
             }
         });
 
+        function skip() {
+            stream.stats.skipped++;
+            stream.stats.done++;
+            // Update length
+            stream.length--;
+            stream.emit('length', stream.length);
+            get(push);
+        }
+
         return true;
     });
 };
-

lib/stream-pyramid.js

@@ -34,6 +34,7 @@ function Pyramid(source, options) {
     this.cursor = undefined;
     this.stats = new Stats();
     this.length = 0;
+    this.job = options.job || false;
 
     stream.Readable.call(this, { objectMode: true });
 }
@@ -49,6 +50,16 @@ Pyramid.prototype._params = function(callback) {
         if (stream.minzoom === undefined) return stream.emit('error', new Error('No minzoom determined'));
         if (stream.maxzoom === undefined) return stream.emit('error', new Error('No maxzoom determined'));
 
+        if (stream.job) {
+            var dx = (stream.bounds[2] - stream.bounds[0]) / stream.job.total;
+            stream.bounds = [
+                stream.bounds[0] + (stream.job.num - 1) * dx,
+                stream.bounds[1],
+                stream.bounds[0] + stream.job.num * dx,
+                stream.bounds[3]
+            ];
+        }
+
         stream.bboxes = {};
         for (var z = stream.minzoom; z <= stream.maxzoom; z++) {
             stream.bboxes[z] = sm.xyz(stream.bounds, z);
@@ -144,4 +155,3 @@ function nextShallow(stream) {
     }
     return true;
 }
-

lib/stream-scanline.js

@@ -30,6 +30,7 @@ function Scanline(source, options) {
     this.bboxes = undefined;
     this.cursor = undefined;
     this.length = 0;
+    this.job = options.job || false;
 
     stream.Readable.call(this, { objectMode: true });
 }
@@ -83,21 +84,30 @@ Scanline.prototype._read = function(size) {
         var x = stream.cursor.x;
         var y = stream.cursor.y;
         nextDeep(stream);
+
+        if (stream.job && x % stream.job.total !== stream.job.num - 1)
+            return skip();
+
         stream.source.getTile(z, x, y, function(err, buffer) {
             if (err && !(/does not exist$/).test(err.message)) {
                 stream.emit('error', err);
             } else if (err || isEmpty(buffer)) {
-                stream.stats.skipped++;
-                stream.stats.done++;
-                // Update length
-                stream.length--;
-                stream.emit('length', stream.length);
-                get(push);
+                skip();
             } else {
                 stream.stats.done++;
                 push(new Tile(z, x, y, buffer));
             }
         });
+
+        function skip() {
+            stream.stats.skipped++;
+            stream.stats.done++;
+            // Update length
+            stream.length--;
+            stream.emit('length', stream.length);
+            get(push);
+        }
+
         return true;
     });
 };
@@ -125,4 +135,3 @@ function nextDeep(stream) {
     }
     return true;
 }
-

lib/stream-util.js

@@ -41,9 +41,11 @@ function serialize(obj) {
     return '';
 }
 
-function deserialize(data) {
+function deserialize(data, property) {
+    if (property) return getSerializedProperty(data, property);
     if (data.indexOf('{"z":') === 0) return deserializeTile(data);
     if (data.indexOf('{') === 0) return deserializeInfo(data);
+
     throw new DeserializationError('Invalid data');
 }
 
@@ -84,6 +86,12 @@ function deserializeTile(data) {
     return new Tile(obj.z, obj.x, obj.y, buf);
 }
 
+function getSerializedProperty(data, property) {
+    var re = new RegExp('"' + property + '":(.+?)[,}]');
+    var m = re.exec(data);
+    return m ? m[1] : undefined;
+}
+
 function deserializeInfo(data) {
     try { obj = JSON.parse(data); }
     catch(err) { throw new DeserializationError('Could not parse data'); }

lib/tilelive.js

@@ -323,4 +323,4 @@ tilelive.streamTypes.list = require('./stream-list');
 var Serialize = require('./stream-serialize');
 var Deserialize = require('./stream-deserialize');
 tilelive.serialize = function() { return new Serialize(); };
-tilelive.deserialize = function() { return new Deserialize(); };
+tilelive.deserialize = function(options) { return new Deserialize(options); };

package.json

@@ -36,12 +36,12 @@
     "sphericalmercator": "~1.0.1"
   },
   "devDependencies": {
-    "tape": "2.13.3",
+    "coveralls": "~2.11.1",
+    "istanbul": "~0.3.0",
     "mbtiles": "~0.4.3",
+    "tape": "2.13.3",
     "tilejson": "~0.8.0",
-    "tilelive-http": "^0.3.0",
-    "istanbul": "~0.3.0",
-    "coveralls": "~2.11.1"
+    "tilelive-http": "^0.3.0"
   },
   "bin": {
     "tilelive-copy": "./bin/tilelive-copy"