Have tile proxy get rendered notifications directly.

TODO

@@ -10,6 +10,4 @@
 
 * commands to force expiration of tiles in queuemaster
 
-* Redo render notifications to hit tile proxy directly
-   - Have renders publish to "toposm.rendered.z.x.y". tp listens to exact
-      topic, queuemaster listens to "toposm.rendered.*".
+* Make renderd max-zoom agnostic

dump_queue.py

@@ -13,7 +13,7 @@
 
 conn = amqp.Connection(host=DB_HOST, userid="guest", password="guest")
 chan = conn.channel()
-chan.exchange_declare(exchange="osm", type="direct", durable=True, auto_delete=False)
+chan.exchange_declare(exchange="osm", type="topic", durable=True, auto_delete=False)
 chan.queue_declare(queue=args.queue, durable=True, exclusive=False, auto_delete=False)
 chan.queue_bind(queue="expire_toposm", exchange="osm", routing_key="expire")
 

expire_tiles.py

@@ -84,7 +84,7 @@ def queue_tile_if_needed(z, x, y, chan, args):
 
     conn = amqp.Connection(host=DB_HOST, userid="guest", password="guest")
     chan = conn.channel()
-    chan.exchange_declare(exchange="osm", type="direct", durable=True, auto_delete=False)
+    chan.exchange_declare(exchange="osm", type="topic", durable=True, auto_delete=False)
 
     expired = read_tiles(args)
     for z in xrange(args.min_zoom, args.max_zoom+1):

queuemaster.py

@@ -351,7 +351,6 @@ def __init__(self, maxz):
         self.expirer = TileExpirer(self.maxz, self.queue)
         self.expirer.start()
         self.renderers = {}
-        self.requests = {}
 
     ### Startup sequence.
 
@@ -366,7 +365,7 @@ def on_connection_open(self, conn):
     def on_channel_open(self, chan):
         self.channel = chan
         chan.exchange_declare(
-            self.on_exchange_declare, exchange="osm", type="direct",
+            self.on_exchange_declare, exchange="osm", type="topic",
             durable=True, auto_delete=False)
 
     def on_exchange_declare(self, frame):
@@ -387,6 +386,9 @@ def on_expire_bind(self,frame):
 
     def on_command_declare(self, frame):
         self.command_queue = frame.method.queue
+        self.channel.queue_bind(
+            None, queue=self.command_queue,
+            exchange='osm', routing_key='toposm.rendered.#')
         self.channel.queue_bind(
             self.on_command_bind, queue=self.command_queue,
             exchange='osm', routing_key='toposm.queuemaster')
@@ -416,7 +418,6 @@ def on_command(self, chan, method, props, body):
                 if props.reply_to in self.renderers:
                     self.remove_renderer(props.reply_to)
             elif command == 'rendered':
-                self.send_render_replies(message['metatile'])
                 if props.reply_to in self.renderers:
                     self.renderers[props.reply_to].finished(message['metatile'])
                 z, x, y = [ int(s) for s in message['metatile'].split('/') ]
@@ -462,25 +463,9 @@ def send_render_requests(self):
         for renderer in self.renderers.values():
             renderer.send_request()
 
-    def send_render_replies(self, mt):
-        if mt in self.requests:
-            for props in self.requests[mt]:
-                self.channel.basic_publish(
-                    exchange='',
-                    routing_key=props.reply_to,
-                    properties=pika.BasicProperties(
-                        correlation_id=props.correlation_id,
-                        content_type='application/json'),
-                    body=json.dumps({'command': 'rendered'}))
-            del self.requests[mt]
-
     def handle_render_request(self, tile, props):
         z, x, y = [ int(i) for i in tile.split('/') ]
         mt = '%s/%s/%s' % (z, x / NTILES[z], y / NTILES[z])
-        if mt in self.requests:
-            self.requests[mt].append(props)
-        else:
-            self.requests[mt] = [props]
         if tileExists(REFERENCE_TILESET, z, x, y):
             importance = 'important'
         else:

renderd.py

@@ -77,7 +77,7 @@ def renderMetaTileFromMsg(self, msg):
         self.printMessage('Notifying queuemaster of completion.')
         self.chan.basic_publish(
             exchange='osm',
-            routing_key='toposm.queuemaster',
+            routing_key='toposm.rendered.{0}.{1}.{2}'.format(z, metax, metay),
             properties=pika.BasicProperties(reply_to=self.commandQueue,
                                             content_type='application/json'),
             body=json.dumps({'command': 'rendered',
@@ -162,7 +162,7 @@ def metaTileNeedsRendering(z, x, y):
 
     conn = pika.BlockingConnection(pika.ConnectionParameters(host=DB_HOST))
     chan = conn.channel()
-    chan.exchange_declare(exchange="osm", type="direct", durable=True, auto_delete=False)
+    chan.exchange_declare(exchange="osm", type="topic", durable=True, auto_delete=False)
     conn.close()
 
     console.printMessage('Starting renderer.')

tp.py

@@ -6,7 +6,6 @@
 import httplib
 import json
 import urllib2
-import uuid
 import socket
 import datetime
 import StringIO
@@ -69,18 +68,16 @@ def render_tile(z, x, y, timeout):
     conn = pika.BlockingConnection(pika.ConnectionParameters(host=DB_HOST))
     chan = conn.channel()
     queue = chan.queue_declare(exclusive=True).method.queue
-    correlation_id = str(uuid.uuid4())
+    chan.queue_bind(queue=queue, exchange='osm', routing_key='toposm.rendered.{0}.{1}.{2}'.format(z, x / NTILES[z], y / NTILES[z]))
     chan.basic_publish(
         exchange='osm',
         routing_key='toposm.queuemaster',
-        properties=pika.BasicProperties(reply_to=queue,
-                                        correlation_id=correlation_id),
         body=json.dumps({'command': 'render',
                          'tile': '{0}/{1}/{2}'.format(z, x, y)}))
     start_time = time.time()
     while time.time() - start_time < timeout:
         (method, props, body) = chan.basic_get(queue=queue, no_ack=True)
-        if method and props.correlation_id == correlation_id:
+        if method:
             return
 
 def render_missing(z, x, y):
@@ -139,7 +136,7 @@ def redirect(z, x, y):
 try:
     z, x, y = [ int(s) for s in os.environ['PATH_INFO'].split('/')[-3:] ]
 
-    sys.stderr.write("{0}/{1}/{2}: {3}\n".format(z, x, y, os.environ['HTTP_USER_AGENT']))
+    sys.stderr.write("request: {0}/{1}/{2}\n".format(z, x, y))
 
     if not tileExists(TILESET[0], z, x, y, TILESET[1]):
         render_missing(z, x, y)