Tornado / Celery / sockets proof of concept

gistfile1.txt

# -*- coding: utf-8 -*-
"""Tornado / Celery / sockets proof of concept
To run:
$ celery -A app worker
$ python app.py
"""

import os
import socket
import random
import string

import celery
import requests
from tornado import web, gen, netutil, ioloop, iostream, httpserver


PORT = 7777
PATH = '/tmp/sock'
URL = 'http://ipv4.download.thinkbroadband.com/200MB.zip'


app = celery.Celery('app')


@app.task
def send_data(path):
    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    client.connect(path)
    resp = requests.get(URL, stream=True)
    for chunk in resp.iter_content(1024*1024):
        client.send(chunk)
    client.close()


class SocketHandler(web.RequestHandler):

    @web.asynchronous
    def get(self):
        name = ''.join([random.choice(string.ascii_lowercase) for _ in range(10)])
        self.path = '/tmp/sock-{}'.format(name)
        sock = netutil.bind_unix_socket(self.path)
        netutil.add_accept_handler(sock, self.handle_connection)
        send_data.delay(self.path)

    def handle_connection(self, s, addr):
        stream = iostream.IOStream(s)
        stream.read_until_close(
            callback=self.handle_close,
            streaming_callback=self.handle_data,
        )

    def handle_data(self, data):
        self.write(data)
        self.flush()
        
    def handle_close(self, data):
        os.remove(self.path)
        self.finish()


if __name__ == '__main__':
    app = web.Application(
        [
            web.url(r'/', SocketHandler),
        ],
        debug=True,
    )
    server = httpserver.HTTPServer(app)
    server.bind(PORT)
    server.start()
    ioloop.IOLoop.current().start()