Merge pull request #13 from trailbehind/fix-write-failures

Fix write failures

README.md

@@ -45,8 +45,8 @@ Usage: node add-tile-list.js [sources]... [options]
 sources     source names to queue
 
 Options:
-   -f FILE, --file FILE   Read list from file. By default list is read from STDIN
-   -v, --version          Show version info
+   -d, --dryRun    Dry run. Don't actually add messages to queue.
+   -v, --version   Show version info
 
 Read a list of tiles to queue for rendering.
 ```

bin/add-tile-list.js

@@ -11,10 +11,10 @@ var nomnom = require("nomnom")
       help: "source names to queue",
       list: true
     },
-    file: {
-      abbr: "f",
-      metavar: "FILE",
-      help: "Read list from file. By default list is read from STDIN"
+    dryRun: {
+      abbr: "d",
+      flag: true,
+      help: "Dry run. Don't actually add messages to queue."
     },
     version: {
       abbr: "v",
@@ -34,7 +34,7 @@ if (!opts.sources || opts.sources.length == 0) {
   process.exit(-1);
 }
 
-new QueueWriter(opts.sources, {}, function(err, queueWriter) {
+new QueueWriter(opts.sources, { dryRun: opts.dryRun }, function(err, queueWriter) {
   var pipe = process.stdin.pipe(require("split")());
   pipe.on("data", function(line) {
     if (line.length) {

lib/queueWriter.js

@@ -15,27 +15,42 @@ function QueueWriter(sources, opts, callback) {
   var tileStream = Writable();
   this.tileStream = tileStream;
   this.dryRun = opts.dryRun || false;
+  var retryDelay = opts.retryDelay || 1000;
   var qw = this;
-
   debug("Connecting to " + server + "/" + ampqTopic);
 
   context.on("ready", function() {
     debug("Connected to context");
     var pub = context.socket("PUSH");
     pub.connect(ampqTopic, function() {
       debug("Connected to socket");
-      tileStream._write = function(chunk, enc, next) {
+      var tilestreamWrite = function(chunk, enc, next, retries) {
+        var success = true;
         for (var sourceIndex = 0; sourceIndex < sources.length; sourceIndex++) {
           var message = sources[sourceIndex] + "+" + chunk;
-          pub.write(message, "utf8");
           debug("Queueing " + message);
+          if (!pub.write(message, "utf8")) {
+            success = false;
+            if (retries === undefined) {
+              retries = 0;
+            }
+            console.log("Error writing to queue, retrying after delay. retry count:" + retries);
+            setTimeout(function() {
+              tilestreamWrite(chunk, enc, next, retries + 1);
+            }, retryDelay);
+            break;
+          }
+        }
+        if (success) {
+          next();
         }
-        next();
       };
+      tileStream._write = tilestreamWrite;
       //Finish needs to be setup here, if it is setup before the socket is connected,
       // and end is called before the connection happens then writes will never happen
       tileStream.on("finish", function() {
         debug("tileStream finished");
+        pub.close();
         setTimeout(function() {
           debug("Exiting");
           process.exit(0);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "tile-squirrel",
-  "version": "1.4.0",
+  "version": "1.5.0",
   "description": "Tile Squirrel reads from a rabbit queue and cache tiles from a tilelive source to a tilelive sink",
   "main": "lib/cacher.js",
   "bin": "bin/squirrel.js",