The asyncore Module

The asyncore module provides a "reactive" socket implementation. Instead of creating socket objects and calling methods on them to do things, this module allows you to write code that is called when something can be done. To implement an asynchronous socket handler, subclass the dispatcher class, and override one or more of the following methods:

• handle_connect is called when a connection is successfully established.
• handle_expt is called when a connection fails.
• handle_accept is called when a connection request is made to a listening socket. The callback should call the accept method to get the client socket.
• handle_read is called when there is data waiting to be read from the socket. The callback should call the recv method to get the data.
• handle_write is called when data can be written to the socket. Use the send method to write data.
• handle_close is called when the socket is closed or reset.
• handle_error(type, value, traceback) is called if a Python error occurs in any of the other callbacks. The default implementation prints an abbreviated traceback to sys.stdout.

Example 7-7 shows a time client, similar to the one for the socket module.

Example 7-7. Using the asyncore Module to Get the Time from a Time Server

File: asyncore-example-1.py

import asyncore
import socket, time

# reference time (in seconds since 1900-01-01 00:00:00)
TIME1970 = 2208988800L # 1970-01-01 00:00:00

class TimeRequest(asyncore.dispatcher):
 # time requestor (as defined in RFC 868)

 def _ _init_ _(self, host, port=37):
 asyncore.dispatcher._ _init_ _(self)
 self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 self.connect((host, port))

 def writable(self):
 return 0 # don't have anything to write

 def handle_connect(self):
 pass # connection succeeded

 def handle_expt(self):
 self.close() # connection failed, shutdown

 def handle_read(self):
 # get local time
 here = int(time.time()) + TIME1970

 # get and unpack server time
 s = self.recv(4)
 there = ord(s[3]) + (ord(s[2])<<8) + (ord(s[1])<<16) + (ord(s[0])<<24L)

 self.adjust_time(int(here - there))

 self.handle_close() # we don't expect more data

 def handle_close(self):
 self.close()

 def adjust_time(self, delta):
 # override this method!
 print "time difference is", delta

#
# try it out

request = TimeRequest("www.python.org")

asyncore.loop()

log: adding channel 
time difference is 28
log: closing channel 192:

If you don't want the log messages, override the log method in your dispatcher subclass.

Example 7-8 shows the corresponding time server. Note that it uses two dispatcher subclasses, one for the listening socket, and one for the client channel.

Example 7-8. Using the asyncore Module to Implement a Time Server

File: asyncore-example-2.py

import asyncore
import socket, time

# reference time
TIME1970 = 2208988800L

class TimeChannel(asyncore.dispatcher):

 def handle_write(self):
 t = int(time.time()) + TIME1970
 t = chr(t>>24&255) + chr(t>>16&255) + chr(t>>8&255) + chr(t&255)
 self.send(t)
 self.close()

class TimeServer(asyncore.dispatcher):

 def _ _init_ _(self, port=37):
 self.port = port
 self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 self.bind(("", port))
 self.listen(5)
 print "listening on port", self.port

 def handle_accept(self):
 channel, addr = self.accept()
 TimeChannel(channel)

server = TimeServer(8037)
asyncore.loop()

log: adding channel 
listening on port 8037
log: adding channel 
log: closing channel 52:

In addition to the plain dispatcher, this module also includes a dispatcher_with_send class. This class allows you send larger amounts of data, without clogging up the network transport buffers.

The module in Example 7-9 defines an AsyncHTTP class based on the dispatcher_with_send class. When you create an instance of this class, it issues an HTTP GET request and sends the incoming data to a "consumer" target object.

Example 7-9. Using the asyncore Module to Do HTTP Requests

File: SimpleAsyncHTTP.py

import asyncore
import string, socket
import StringIO
import mimetools, urlparse

class AsyncHTTP(asyncore.dispatcher_with_send):
 # HTTP requester

 def _ _init_ _(self, uri, consumer):
 asyncore.dispatcher_with_send._ _init_ _(self)

 self.uri = uri
 self.consumer = consumer

 # turn the uri into a valid request
 scheme, host, path, params, query, fragment = urlparse.urlparse(uri)
 assert scheme == "http", "only supports HTTP requests"
 try:
 host, port = string.split(host, ":", 1)
 port = int(port)
 except (TypeError, ValueError):
 port = 80 # default port
 if not path:
 path = "/"
 if params:
 path = path + ";" + params
 if query:
 path = path + "?" + query

 self.request = "GET %s HTTP/1.0
Host: %s

" % (path, host)

 self.host = host
 self.port = port

 self.status = None
 self.header = None

 self.data = ""

 # get things going!
 self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 self.connect((host, port))

 def handle_connect(self):
 # connection succeeded
 self.send(self.request)

 def handle_expt(self):
 # connection failed; notify consumer (status is None)
 self.close()
 try:
 http_header = self.consumer.http_header
 except AttributeError:
 pass
 else:
 http_header(self)

 def handle_read(self):
 data = self.recv(2048)
 if not self.header:
 self.data = self.data + data
 try:
 i = string.index(self.data, "

")
 except ValueError:
 return # continue
 else:
 # parse header
 fp = StringIO.StringIO(self.data[:i+4])
 # status line is "HTTP/version status message"
 status = fp.readline()
 self.status = string.split(status, " ", 2)
 # followed by a rfc822-style message header
 self.header = mimetools.Message(fp)
 # followed by a newline, and the payload (if any)
 data = self.data[i+4:]
 self.data = ""
 # notify consumer (status is non-zero)
 try:
 http_header = self.consumer.http_header
 except AttributeError:
 pass
 else:
 http_header(self)
 if not self.connected:
 return # channel was closed by consumer

 self.consumer.feed(data)

 def handle_close(self):
 self.consumer.close()
 self.close()

Example 7-10 shows a simple script that uses that class.

Example 7-10. Using the SimpleAsyncHTTP Class

File: asyncore-example-3.py

import SimpleAsyncHTTP
import asyncore

class DummyConsumer:
 size = 0

 def http_header(self, request):
 # handle header
 if request.status is None:
 print "connection failed"
 else:
 print "status", "=>", request.status
 for key, value in request.header.items():
 print key, "=", value

 def feed(self, data):
 # handle incoming data
 self.size = self.size + len(data)

 def close(self):
 # end of data
 print self.size, "bytes in body"

#
# try it out

consumer = DummyConsumer()

request = SimpleAsyncHTTP.AsyncHTTP(
 "http://www.pythonware.com",
 consumer
 )

asyncore.loop()

log: adding channel 
status => ['HTTP/1.1', '200', 'OK1512']
server = Apache/Unix (Unix)
content-type = text/html
content-length = 3730
...
3730 bytes in body
log: closing channel 156:

Note that the consumer interface is designed to be compatible with the htmllib and xmllib parsers, allowing you to parse HTML or XML data on the fly. Note that the http_header method is optional; if it isn't defined, it's simply ignored.

A problem with Example 7-10 is that it doesn't work for redirected resources. Example 7-11 adds an extra consumer layer, which handles the redirection.

Example 7-11. Using the SimpleAsyncHTTP Class with Redirection

File: asyncore-example-4.py

import SimpleAsyncHTTP
import asyncore

class DummyConsumer:
 size = 0

 def http_header(self, request):
 # handle header
 if request.status is None:
 print "connection failed"
 else:
 print "status", "=>", request.status
 for key, value in request.header.items():
 print key, "=", value

 def feed(self, data):
 # handle incoming data
 self.size = self.size + len(data)

 def close(self):
 # end of data
 print self.size, "bytes in body"

class RedirectingConsumer:

 def _ _init_ _(self, consumer):
 self.consumer = consumer

 def http_header(self, request):
 # handle header
 if request.status is None or
 request.status[1] not in ("301", "302"):
 try:
 http_header = self.consumer.http_header
 except AttributeError:
 pass
 else:
 return http_header(request)
 else:
 # redirect!
 uri = request.header["location"]
 print "redirecting to", uri, "..."
 request.close()
 SimpleAsyncHTTP.AsyncHTTP(uri, self)

 def feed(self, data):
 self.consumer.feed(data)

 def close(self):
 self.consumer.close()

#
# try it out

consumer = RedirectingConsumer(DummyConsumer())

request = SimpleAsyncHTTP.AsyncHTTP(
 "http://www.pythonware.com/library",
 consumer
 )

asyncore.loop()

log: adding channel 
redirecting to http://www.pythonware.com/library/ ...
log: closing channel 48:
log: adding channel 
status => ['HTTP/1.1', '200', 'OK1512']
server = Apache/Unix (Unix)
content-type = text/html
content-length = 387
...
387 bytes in body
log: closing channel 236:

If the server returns status 301 (permanent redirection) or 302 (temporary redirection), the redirecting consumer closes the current request and then issues a new one for the new address. All other calls to the consumer are delegated to the original consumer.