Little dumb experiment with Tornado and 0MQ

tornado-zmq.py

# This was a little experiment I've put together in some minutes while in TDC2011 (The Developers Conference), at São Paulo
# It's just a silly almost-Hello-World, just to show the assynchronous communication working with non-blocking I/O through event loops, mixed with a simple HTTP server

# First, put this on a file called 'server.py', and run:
#$ ./server.py 8001 &
#$ ./server.py 8002 &
#$ ./server.py 8003 &


#!/usr/bin/env python

import random
import sys
import time

import zmq
from zmq.eventloop import IOLoop

loop = IOLoop.instance()

context = zmq.Context()

rep = context.socket(zmq.REP)
port = sys.argv[1]
rep.bind("tcp://127.0.0.1:%s" % port)

def on_receive(sock, events):
    sock.recv()
    time.sleep(random.randint(10, 100) / 1000.0)
    sock.send(port)

loop.add_handler(rep, on_receive, zmq.POLLIN)

loop.start()




# Now, put this on a file called 'client.py', and run:
#$ ./client.py 8001 8002 8003


#!/usr/bin/env python

import sys

import tornado
import tornado.web
import zmq
from zmq.eventloop import ioloop, zmqstream

tornado.ioloop = ioloop


class Handler(tornado.web.RequestHandler):
    def __init__(self, *args, **kwargs):
        super(Handler, self).__init__(*args, **kwargs)

        self.counter = 0
        self.requests = 10
        self.streams = []

        self.setup_sockets()

    def setup_sockets(self):
        context = zmq.Context()

        ports = sys.argv[1:]
        for port in ports:
            self.create_streams(context, port)

    def create_streams(self, context, port):
        req = context.socket(zmq.REQ)
        req.connect('tcp://127.0.0.1:%s' % port)
        stream = zmqstream.ZMQStream(req, tornado.ioloop.IOLoop.instance())
        stream.on_recv(self.receive)
        self.streams.append(stream)

    def receive(self, port):
        self.write('response from %s\n' % port)
        self.counter += 1
        if self.received_all_responses():
            self.finish()

    def received_all_responses(self):
        return self.counter == (len(self.streams) * self.requests)

    @tornado.web.asynchronous
    def get(self):
        for i in range(10):
            for stream in self.streams:
                stream.send("")

application = tornado.web.Application([(r'/', Handler)])

application.listen(8888)

tornado.ioloop.IOLoop.instance().start()




# Now, see how the random processing times from the services respond back to the client, asynchronously, and the client prints the order in which it receives these replies in the HTTP response:
#$ curl http://127.0.0.1:8888

# What happened?
#The client started an async communication channel with each server. Then, when the client receives an HTTP request, it sends a message to each of the servers, and defines a callback to receive the response from these messages. On each message to a server, this sleeps for a random time between 0.01 and 0.1 seconds, and sends back a message with the data being the port number it's running at.
# On each message the client receives, it writes a message with the port from the related server in the (still open) HTTP response it's supposed to send to the other client (curl) that made the HTTP request.
# Then, it closes the HTTP response, closes the HTTP request and sends the response to curl. :-)

This is now at: https://github.com/diogobaeder/tornado-zmq-lab