tornado-based async POP3 client

protocols.py

class CommandFailedException(Exception):
    def __init__(self, result=None):
        Exception.__init__(self)
        self.result = result
        if result:
            self.message = "Mail command failed: \"%s\"" % result

class AsyncMailClient(object):
    def __init__(self, host, port, use_ssl=True, timeout=4):
        self.host = host
        self.port = port
        self.use_ssl = use_ssl
        self.timeout = timeout
        self.stream = None
        
    def connect(self, callback):
        """ Connects to server. Callback receives welcome line. """
        if not self.stream or self.stream.closed():
            def connect_callback():
                self._read_line(callback)
            
            self.stream = (SSLIOStream if self.use_ssl else IOStream)(socket.socket())
            self.stream.connect((self.host, self.port), 
                                connect_callback)

class AsyncPOP3Client(AsyncMailClient):
    """ Non-blocking POP3 client, almost compatible with poplib POP3.
        This class also handles POP3 connections requiring SSL (use_ssl 
        parameter of __init__ or port set to 995).
    """
      
    def __init__(self, host, port=110, use_ssl=None, timeout=4):
        use_ssl = use_ssl if use_ssl is not None else (port == 995)
        AsyncMailClient.__init__(self, host, port, use_ssl, timeout)
    
    def user(self, username, callback):
        """ Send username to server """
        def callback_():
            self._read_line(callback)
        self.stream.write("user %s\r\n" % username, callback_)
    
    def pass_(self, password, callback):
        """ Send *plaintext* password to server """
        def callback_():
            self._read_line(callback)
        self.stream.write("pass %s\r\n" % password, callback_)
    
    def stat(self, callback):
        """ Get mailbox status. Result in form (message count, mailbox size). """
        def write_callback():
            self._read_line(callback_)
            
        def callback_(data):
            data = data.split(" ")
            callback((int(data[0]), int(data[1])))
        
        self.stream.write("stat\r\n", write_callback)   
    
    def list(self, callback, which=None):
        def write_callback():
            self._read_long(callback_)
        
        def callback_(data):
            callback(self._parse_long(data))
            
        if which:
            self.stream.write("list %d\r\n" % which, write_callback)
        else:
            self.stream.write("list\r\n", write_callback)        
    
    def retr(self, which, callback):
        """ Retrieve given message. 
            Result in form (response, [ 'line', ...], octets)
        """
        
        def write_callback():
            self._read_long(callback_)
        
        def callback_(data):
            callback(self._parse_long(data))
        
        self.stream.write("retr %d\r\n" % which, write_callback)
    
    def dele(self, which, callback):
        """ Delete given message. Message is not deleted from server until 
            client issues QUIT command.
        """
        
        def callback_():
            self._read_line(callback)
        self.stream.write("dele %d\r\n" % which, callback_)
        
    def rset(self, callback):
        """ Void all deletion marks """
        def callback_():
            self._read_line(callback)
        self.stream.write("rset\r\n", callback_)
        
    def noop(self, callback):
        """ Issue a NOOP command (useful for keeping connection alive). """
        def callback_():
            self._read_line(callback)
        self.stream.write("noop\r\n", callback_)
    
    def quit(self, callback):
        """ Quit (all messages marked for deletion will be deleted). """
        def callback_():
            # this looks bad and I have no idea how to change it
            def read_line_callback(data):
                callback(data)
                self.stream.close()
            
            self._read_line(read_line_callback)
        self.stream.write("quit\r\n", callback_)
    
    def top(self, which, callback, howmuch=0):
        """ Receive message's headers and howmuch lines from it's body.
            Result in form (response, [ 'line', ...], octets).
        """
        def write_callback():
            self._read_long(callback_)
        
        def callback_(data):
            callback(self._parse_long(data))
        
        self.stream.write("top %d %d\r\n" % (which, howmuch), write_callback)
    
    def uidl(self, callback, which=None):
        def write_callback():
            self._read_long(callback_)
        
        def callback_(data):
            callback(self._parse_long(data))
        
        if which:
            self.stream.write("uidl %d\r\n" % which, write_callback)
        else:
            self.stream.write("uidl\r\n", write_callback)      
    
    def _parse_long(self, data):
        lines = data.split("\r\n")
        response = lines.pop(0)
        del lines[-3:] # delete \r\n.\r\n part
        return (response, lines, len(data) - len(response) - 5)
    
    def _read_line(self, unwrapped_callback):
        self.stream.read_until("\r\n", self._wrap_callback(unwrapped_callback))
        
    def _read_long(self, unwrapped_callback):
        self.stream.read_until("\r\n.\r\n", 
                               self._wrap_callback(unwrapped_callback))
    
    def _wrap_callback(self, callback):
        @wraps(callback)
        def wrapper(data, *args, **kwargs):
            if data[0:3] != '+OK':
                raise CommandFailedException(data)
            return callback(data, *args, **kwargs)
        return wrapper