configurable timeout interval, one min default

lib/tilelive.js

@@ -342,7 +342,7 @@ tilelive.copy = function(src, dst, options, callback) {
     q.await(function(err) {
         if (err) return callback(err);
 
-        copy(src, dst, options, function(err) {
+            copy(src, dst, options, function(err) {
             if (err) return callback(err);
             if (options.close) closingTime(src, dst, function(err) {
                 if (err) throw err;
@@ -360,22 +360,28 @@ tilelive.copy = function(src, dst, options, callback) {
         var put = sinkStream ?
             options.outStream : tilelive.createWriteStream(dst, {retry:opts.retry});
 
-        var hasErrored = false;
-        function handleError(err) {
-            if (!hasErrored) callback(err);
-            hasErrored = true;
+        var sentCallback = false;
+        function done(err) {
+            if (!sentCallback) {
+                clearInterval(timeout);
+                callback(err);
+            }
+            sentCallback = true;
         }
 
         var previous;
-        setInterval(function() {
+        var timeout = setInterval(function() {
             if (prog.progress.transferred === previous) {
-                // kill somehow
+                done(new Error('Copy operation timed out'));
+                pipeline.unpipe(prog);
+                prog.end();
+                clearInterval(timeout);
             }
             previous = prog.progress.transferred;
-        }, 60000).unref();
+        }, opts.timeout || 60000);
 
-        get.on('error', handleError);
-        put.on('error', handleError);
+        get.on('error', done);
+        put.on('error', done);
         get.on('length', prog.setLength);
 
         if (options.progress)
@@ -384,9 +390,9 @@ tilelive.copy = function(src, dst, options, callback) {
         var doneEvent = sinkStream ? 'finish' : 'stop';
 
         if (options.outStream === process.stdout ||
-            options.outStream === process.stderr) prog.on('end', callback);
+            options.outStream === process.stderr) prog.on('end', done);
         else
-            put.on(doneEvent, callback);
+            put.on(doneEvent, done);
 
         var pipeline = opts.type === 'list' ? opts.listStream.pipe(get) : get;
         if (sinkStream) pipeline = pipeline.pipe(tilelive.serialize());

test/copy.test.js

@@ -148,16 +148,16 @@ test('tilelive copy: missing liststream', function(t) {
 test('tilelive.copy: close src/dst', function(t) {
     var src = new Timedsource({});
     src.flag = false;
-    src.close = function(callback) { 
-        this.flag = true; 
-        callback(); 
+    src.close = function(callback) {
+        this.flag = true;
+        callback();
     };
 
     var dst = new Timedsource({});
     dst.flag = false;
-    dst.close = function(callback) { 
-        this.flag = true; 
-        callback(); 
+    dst.close = function(callback) {
+        this.flag = true;
+        callback();
     };
 
     var options = {
@@ -307,6 +307,17 @@ test('tilelive.copy + write err (retry)', function(t) {
     });
 });
 
+test('tilelive.copy timeout', function(t) {
+    var src = new Timedsource({timeout: 1000});
+    var dst = new Timedsource({});
+    var options = { timeout: 1000, maxzoom: 21 };
+    tilelive.copy(src, dst, options, function(err) {
+        t.ok(err, 'expected error message');
+        t.equal(err.message, 'Copy operation timed out', 'timeout error');
+        t.end();
+    });
+});
+
 // Used for progress report
 function report(stats, p) {
     util.print(util.format('\r\033[K[%s] %s%% %s/%s @ %s/s | ✓ %s □ %s | %s left',

test/stream-scanline.test.js

@@ -192,7 +192,7 @@ test('scanline: invalid extent', function(assert) {
     require('../lib/stream-util').retryBackoff = 1;
     var get = tilelive.createReadStream(fakesrc, {type:'scanline'});
     var put = tilelive.createWriteStream(new Timedsource({}));
-    get.on('error', function(err) {         
+    get.on('error', function(err) {
         assert.equal(err.message, 'bounds must be an array of the form [west, south, east, north]');
     });
     get.pipe(put);

test/timedsource.js

@@ -20,6 +20,13 @@ function Timedsource(uri, callback) {
     this.fail = uri.fail || 0;
     this.fails = {};
 
+    if (uri.timeout) {
+        var timedsource = this;
+        setTimeout(function() {
+            timedsource.hang = true;
+        }, uri.timeout);
+    }
+
     if (callback) callback(null, this);
     return this;
 }
@@ -45,6 +52,8 @@ Timedsource.prototype.getTile = function(z, x, y, callback) {
         fails[key] = fails[key] || 0;
     }
 
+    if (this.hang) return;
+
     setTimeout(function() {
         if (fail && fails[key] < fail) {
             fails[key]++;