FND · August 30, 2024 04:42 · Jun 4, 2015 · Jun 4, 2015
diff --git a/aggregator.py b/aggregator.py
@@ -2,12 +2,14 @@
 from random import randint
 
 
-def retrieve(*uris):
+def retrieve(*uris, callback=None):
     res = Future()
     responses = []
 
     def conclude(response):
         responses.append(response)
+        if callback:
+            callback(response)
         if len(responses) == len(uris):
             res.set_result(responses)
 
@@ -24,7 +26,7 @@ def fetch(uri):
     lag = yield from _http()
     lag = lag * 1000
     print("downloaded %s in %d ms" % (uri, lag))
-    return "[%d ms] %s" % (lag, uri)
+    return "[%d ms] %s\n" % (lag, uri)
 
 
 @async

diff --git a/app.py b/app.py
@@ -10,13 +10,32 @@
 class Frontpage:
 
     def on_get(self, req, res):
-        service_requests = aggregator.retrieve("http://example.org/foo",
-                "http://example.org/bar", "http://example.org/baz")
+        uris = ("http://example.org/foo", "http://example.org/bar",
+                "http://example.org/baz")
+
+        res.stream = ResponseStream(res.stream)
+        service_requests = aggregator.retrieve(*uris,
+                callback=res.stream.write) # XXX: race condition? `write` vs. `__iter__`
         loop = asyncio.get_event_loop()
         loop.run_until_complete(service_requests)
 
-        results = service_requests.result()
-        res.body = "%s\n" % "\n".join(results)
+
+class ResponseStream:
+
+    def __init__(self, output_stream):
+        self.output_stream = output_stream
+        self.chunks = []
+
+    def write(self, chunk):
+        self.chunks.append(chunk)
+
+    def __iter__(self, *args, **kwargs):
+        while True:
+            try:
+                chunk = self.chunks.pop(0) # XXX: inefficient?
+                yield chunk.encode("utf8")
+            except IndexError:
+                break
 
 
 app = falcon.API()

diff --git a/README.md b/README.md
@@ -0,0 +1,10 @@
+Getting Started
+---------------
+
+    $ pip install falcon
+    $ python3 app.py
+
+alternatively with Gunicorn (for HTTP/1.1):
+
+    $ pip install gunicorn
+    $ gunicorn app:app
diff --git a/aggregator.py b/aggregator.py
@@ -0,0 +1,34 @@
+from asyncio import Future, Task, coroutine as async, sleep
+from random import randint
+
+
+def retrieve(*uris):
+    res = Future()
+    responses = []
+
+    def conclude(response):
+        responses.append(response)
+        if len(responses) == len(uris):
+            res.set_result(responses)
+
+    for uri in uris:
+        task = Task(fetch(uri))
+        task.add_done_callback(lambda task: conclude(task.result()))
+
+    return res
+
+
+@async
+def fetch(uri):
+    print("retrieving %s..." % uri)
+    lag = yield from _http()
+    lag = lag * 1000
+    print("downloaded %s in %d ms" % (uri, lag))
+    return "[%d ms] %s" % (lag, uri)
+
+
+@async
+def _http():
+    lag = randint(2, 9) * 0.1
+    yield from sleep(lag)
+    return lag
diff --git a/app.py b/app.py
@@ -0,0 +1,28 @@
+import asyncio
+
+import falcon
+
+import aggregator
+
+from wsgiref import simple_server
+
+
+class Frontpage:
+
+    def on_get(self, req, res):
+        service_requests = aggregator.retrieve("http://example.org/foo",
+                "http://example.org/bar", "http://example.org/baz")
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(service_requests)
+
+        results = service_requests.result()
+        res.body = "%s\n" % "\n".join(results)
+
+
+app = falcon.API()
+app.add_route("/", Frontpage())
+
+
+if __name__ == "__main__":
+    server = simple_server.make_server("localhost", 8000, app)
+    server.serve_forever()
No results found