txwsgi / doc / examples /

# Copyright (c) 2010 Manlio Perillo (
# See LICENSE for details.

import md5
from cgi import parse_qs

from twisted.internet import reactor
from twisted.application import service, strports
from twisted.web import server

from txwsgi import WSGIResource

BUFSIZE = 4096
DELAY = 10 # ms

def wsgi_application(environ, start_response):
    """A WSGI application that compute the MD5 hash of the file
    specified in the PATH_INFO.

    I{x-wsgiorg.suspend} extension is used to yield control to HTTP
    server, after each chunk is processed.

    Buffer size to use can be specified using the I{bufsize} parameter
    in the query string.

    Timeout to use in I{x-wsgiorg.suspend} can be specified using the
    I{delay} parameter in the query string.

    def app_iter():
        hash =
        while True:
            data =
            if not data:


            yield ''

        yield 'Test hash %s' % hash.hexdigest()

    headers = [('Content-Type', 'text/plain')]
    suspend = environ['x-wsgiorg.suspend']

    # Parse query string
    qs = environ.get('QUERY_STRING', '')
    args = parse_qs(qs)
    bufsize = int(args.get('bufsize', [BUFSIZE])[0])
    delay = int(args.get('delay', [DELAY])[0])

    # Resolve path info to filesystem path, relative to the root
    # directory.
    # Only work on POSIX systems.
    path = '/' + environ['PATH_INFO']
        fp = file(path)
    except IOError:
        start_response('404 NOT FOUND', headers)
        return ['path "%s" not found' % path]

    start_response('200 OK', headers)
    return app_iter()

res = WSGIResource(reactor, wsgi_application)
site = server.Site(res)
application = service.Application('demo cooperative')

s = strports.service('tcp:8080', site)