added wsgitools.scgi.forkpool

wsgitools.scgi is now known as wsgitools.scgi.asynchronous

1 files changed, 201 insertions, 0 deletions

diff --git a/wsgitools/scgi/asynchronous.py b/wsgitools/scgi/asynchronous.py
new file mode 100644
index 0000000..51349a0
--- /dev/null
+++ b/wsgitools/scgi/asynchronous.py
@@ -0,0 +1,201 @@
+__all__ = []
+
+import asyncore
+import socket
+import sys
+try:
+    import cStringIO as StringIO
+except ImportError:
+    import StringIO
+
+class SCGIConnection(asyncore.dispatcher):
+    """SCGI connection class used by WSGISCGIServer."""
+    # maximum request size
+    MAX_REQUEST_SIZE = 65536
+    # maximum post size
+    MAX_POST_SIZE = 8 << 20
+    # read and write size
+    BLOCK_SIZE = 4096
+    # connection states
+    NEW    = 0*4 | 1 # connection established, waiting for request
+    HEADER = 1*4 | 1 # the request length was received, waiting for the rest
+    BODY   = 2*4 | 1 # the request header was received, waiting for the body
+    REQ    = 3*4 | 2 # request received, sending response
+    def __init__(self, server, connection, addr):
+        self.server = server # WSGISCGIServer instance
+        self.addr = addr # scgi client address
+        self.state = SCGIConnection.NEW # internal state
+        self.environ = {} # environment passed to wsgi app
+        self.reqlen = -1 # request length used in two different meanings
+        self.inbuff = "" # input buffer
+        self.outbuff = "" # output buffer
+        self.wsgihandler = None # wsgi application iterator
+        self.outheaders = () # headers to be sent
+                             # () -> unset, (..,..) -> set, True -> sent
+        self.body = StringIO.StringIO() # request body
+        asyncore.dispatcher.__init__(self, connection)
+
+    def _wsgi_headers(self):
+        return {"wsgi.version": (1, 0),
+                "wsgi.input": self.body,
+                "wsgi.errors": self.server.error,
+                "wsgi.url_scheme": "http", # TODO: this is wrong
+                "wsgi.multithread": False,
+                "wsgi.multiprocess": False,
+                "wsgi.run_once": False}
+
+    def _try_send_headers(self):
+        if self.outheaders != True:
+            assert not self.outbuff
+            status, headers = self.outheaders
+            headdata = "".join(map("%s: %s\r\n".__mod__, headers))
+            self.outbuff = "Status: %s\r\n%s\r\n" % (status, headdata)
+            self.outheaders = True
+
+    def _wsgi_write(self, data):
+        assert self.state >= SCGIConnection.REQ
+        if data:
+            self._try_send_headers()
+            self.outbuff += data
+
+    def readable(self):
+        """asyncore interface"""
+        return self.state & 1 == 1
+
+    def writable(self):
+        """asyncore interface"""
+        return self.state & 2 == 2
+
+    def handle_read(self):
+        """asyncore interface"""
+        data = self.recv(self.BLOCK_SIZE)
+        self.inbuff += data
+        if self.state == SCGIConnection.NEW:
+            if ':' in self.inbuff:
+                reqlen, self.inbuff = self.inbuff.split(':', 1)
+                if not reqlen.isdigit():
+                    self.close()
+                    return # invalid request format
+                reqlen = long(reqlen)
+                if reqlen > self.MAX_REQUEST_SIZE:
+                    self.close()
+                    return # request too long
+                self.reqlen = reqlen
+                self.state = SCGIConnection.HEADER
+            elif len(self.inbuff) > self.MAX_REQUEST_SIZE:
+                self.close()
+                return # request too long
+
+        if self.state == SCGIConnection.HEADER:
+            buff = self.inbuff[:self.reqlen]
+            remainder = self.inbuff[self.reqlen:]
+
+            while buff.count('\0') >= 2:
+                key, value, buff = buff.split('\0', 2)
+                self.environ[key] = value
+                self.reqlen -= len(key) + len(value) + 2
+
+            self.inbuff = buff + remainder
+
+            if self.reqlen == 0:
+                if self.inbuff.startswith(','):
+                    self.inbuff = self.inbuff[1:]
+                    self.reqlen = long(self.environ["CONTENT_LENGTH"])
+                    if self.reqlen > self.MAX_POST_SIZE:
+                        self.close()
+                        return
+                    self.state = SCGIConnection.BODY
+                else:
+                    self.close()
+                    return # protocol violation
+
+        if self.state == SCGIConnection.BODY:
+            if len(self.inbuff) >= self.reqlen:
+                self.body.write(self.inbuff[:self.reqlen])
+                self.body.seek(0)
+                self.inbuff = ""
+                self.reqlen = 0
+                self.environ.update(self._wsgi_headers())
+                if "HTTP_CONTENT_TYPE" in self.environ:
+                    self.environ["CONTENT_TYPE"] = \
+                                self.environ.pop("HTTP_CONTENT_TYPE")
+                if "HTTP_CONTENT_LENGTH" in self.environ:
+                    del self.environ["HTTP_CONTENT_LENGTH"] # TODO: better way?
+                self.wsgihandler = iter(self.server.wsgiapp(
+                                self.environ, self.start_response))
+                self.state = SCGIConnection.REQ
+            else:
+                self.body.write(self.inbuff)
+                self.reqlen -= len(self.inbuff)
+                self.inbuff = ""
+
+    def start_response(self, status, headers, exc_info=None):
+        if exc_info:
+            if self.outheaders == True:
+                try:
+                    raise exc_info[0], exc_info[1], exc_info[2]
+                finally:
+                    exc_info = None
+        assert self.outheaders != True # unsent
+        self.outheaders = (status, headers)
+        return self._wsgi_write
+
+    def handle_write(self):
+        """asyncore interface"""
+        assert self.state >= SCGIConnection.REQ
+        if len(self.outbuff) < self.BLOCK_SIZE:
+            for data in self.wsgihandler:
+                self.outbuff += data
+                if len(self.outbuff) >= self.BLOCK_SIZE:
+                    self._try_send_headers()
+                    break
+            if len(self.outbuff) == 0:
+                if hasattr(self.wsgihandler, "close"):
+                    self.wsgihandler.close()
+                self.close()
+                return
+        try:
+            sentbytes = self.send(self.outbuff[:self.BLOCK_SIZE])
+        except socket.error:
+            if hasattr(self.wsgihandler, "close"):
+                self.wsgihandler.close()
+            self.close()
+            return
+        self.outbuff = self.outbuff[sentbytes:]
+
+    def handle_close(self):
+        """asyncore interface"""
+        self.close()
+
+__all__.append("SCGIServer")
+class SCGIServer(asyncore.dispatcher):
+    """SCGI Server for WSGI applications. It does not use multiple processes or
+    multiple threads."""
+    def __init__(self, wsgiapp, port, interface="localhost", error=sys.stderr):
+        """wsgiapp is the wsgi application to be run.
+        port is an int representing the TCP port number to be used.
+        interface is a string specifying the network interface to bind which
+                  defaults to "localhost" making the server inaccessible over
+                  network.
+        error is a file-like object being passed as wsgi.error in the environ
+              parameter defaulting to stderr."""
+        asyncore.dispatcher.__init__(self)
+        self.wsgiapp = wsgiapp
+        self.error = error
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.set_reuse_addr()
+        self.bind((interface, port))
+        self.listen(5)
+
+    def handle_accept(self):
+        """asyncore interface"""
+        ret = self.accept()
+        if ret is not None:
+            conn, addr = ret
+            SCGIConnection(self, conn, addr)
+
+    def run(self):
+        """Runs the server. It will not return and you can invoke
+        asyncore.loop() instead achieving the same effect."""
+        asyncore.loop()
+