Merge lp:~cmiller/desktopcouch/kill-asyncore-use-twisted into lp:desktopcouch

[Chad Miller]

[r=statik] Change pairing tool to twisted instead of asyncore.

[Chad Miller (cmiller)]

I, for one, welcome our new twisted overlords.

[Elliot Murphy (statik)]

Looks good! please file a bug for the couple of FIXME spots.

review: Approve

=== modified file 'desktopcouch/pair/couchdb_pairing/dbus_io.py'
--- desktopcouch/pair/couchdb_pairing/dbus_io.py	2009-07-08 17:48:11 +0000
+++ desktopcouch/pair/couchdb_pairing/dbus_io.py	2009-07-10 22:13:01 +0000
@@ -26,7 +26,7 @@
 
 
 def get_local_hostname():
-    """Get the name of this host, an ASCII-encoded string."""
+    """Get the name of this host, as Unicode host and domain parts."""
     bus, server = get_dbus_bus_server()
     return server.GetHostName(), server.GetDomainName()
 

=== modified file 'desktopcouch/pair/couchdb_pairing/network_io.py'
--- desktopcouch/pair/couchdb_pairing/network_io.py	2009-07-08 17:48:11 +0000
+++ desktopcouch/pair/couchdb_pairing/network_io.py	2009-07-10 22:13:01 +0000
@@ -15,11 +15,14 @@
 # along with desktopcouch.  If not, see <http://www.gnu.org/licenses/>.
 """All inter-tool communication."""
 
-import asyncore
-import socket
 import logging
 import hashlib
 
+from twisted.internet import reactor
+from twisted.internet.protocol import ServerFactory, ReconnectingClientFactory
+from twisted.protocols import basic
+from twisted.internet import ssl
+
 import dbus
 
 try:
@@ -29,132 +32,81 @@
     get_remote_hostname = lambda addr: None
     
 
-# asyncore is ancient.   pylint: disable-msg=W0233
-# asyncore overridden.   pylint: disable-msg=C0111
-
-message_terminator = "\n"
 hash = hashlib.sha512
 
-def receive_up_to_char(s, buffer_list, sentinel=message_terminator):
-    message = s.recv(4098)
-    if len(message) == 0:
-        return None
-
-    buffer_list.append(message)
-    total_so_far = "".join(buffer_list)
-    if sentinel not in total_so_far:
-        return None
-
-    message, extra = total_so_far.split(sentinel, 1)
-    buffer_list[:] = [ extra ]
-
-    return message
-
-
-
-class ListenForInvitations(asyncore.dispatcher):
+
+class ListenForInvitations():
     """Narrative "Alice".
 
     This is the first half of a TCP listening socket.  We spawn off
     processors when we accept invitation-connections."""
 
-    def __init__(self, get_secret_from_user, on_close, local_map=False):
-        self.logging = logging.getLogger("ListenForInvitations")
-
-        if local_map:
-            if not hasattr(local_map, "get"):
-                local_map = dict()
-            self.__map = local_map
-            asyncore.dispatcher.__init__(self, map=self.__map)
-        else:
-            self.__map = None
-            asyncore.dispatcher.__init__(self)
-
-        self.get_secret_from_user = get_secret_from_user
-        self.on_close = on_close
-
-        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.bind(("", 0))
-        self.listen(5)
-
-    def process(self):
-        """How we jump into asyncore processing.  Call this occasionally."""
-        asyncore.loop(0.01, False, self.__map, 3)
-        return True
+    def __init__(self, get_secret_from_user, on_close):
+        """Initialize."""
+        self.logging = logging.getLogger(self.__class__.__name__)
+
+        self.factory = ProcessAnInvitationFactory(get_secret_from_user,
+                on_close)
+        self.listening_port = reactor.listenTCP(0, self.factory)
 
     def get_local_port(self):
         """We created a socket, and the caller needs to know what our port
         number is, so it can advertise it."""
 
-        bogus_address, port = self.getsockname()
+        port = self.listening_port.getHost().port
         self.logging.info("local port to receive invitations is %s", port)
         return port
 
-    def handle_accept(self):
-        s, remote_addr = self.accept()
-        self.logging.debug("accepted connection from %s", s)
-        ProcessAnInvitation(self.get_secret_from_user, s, self.on_close,
-                local_map=self.__map)
-
-
-class ProcessAnInvitation(asyncore.dispatcher_with_send):
+    def close(self):
+        """Called from the UI when a window is destroyed and we do not need
+        this connection any more."""
+        self.listening_port.stopListening()
+
+
+class ProcessAnInvitationProtocol(basic.LineReceiver):
     """Narrative "Alice".
 
     Listen for messages, and when we receive one, call the display callback
     function with the inviter details plus a key."""
 
-    def __init__(self, get_secret_from_user, sock, on_close, local_map=False):
-        self.logging = logging.getLogger("ProcessAnInvitation")
-        
-        if local_map:
-            assert hasattr(local_map, "get")
-            asyncore.dispatcher_with_send.__init__(self, sock, map=local_map)
-        else:
-            asyncore.dispatcher_with_send.__init__(self, sock)
-
+    def __init__(self):
+        """Initialize."""
+        self.logging = logging.getLogger(self.__class__.__name__)
         self.expected_hash = None
         self.public_seed = None
 
-        self.get_secret_from_user = get_secret_from_user
-        self.on_close = on_close
-        self.recv_buffers = []
-        self.logging.debug("initialized")
-
-    def handle_connect(self):
-        self.logging.debug("connecting")
-
-    def handle_close(self):
-        self.logging.debug("closing")
-        self.on_close()
-        self.close()
-
-    def handle_read(self):
-        message = receive_up_to_char(self, self.recv_buffers)
-        if message is None:
-            return
-
+    def connectionMade(self):
+        """Called when a connection is made.  No obligation here."""
+        basic.LineReceiver.connectionMade(self)
+
+    def connectionLost(self):
+        """Called when a connection is lost."""
+        self.logging.debug("connection lost")
+        basic.LineReceiver.connectionLost(self)
+
+    def lineReceived(self, message):
+        """Handler for receipt of a message from the Bob end."""
         h = hash()
         digest_nybble_count = h.digest_size * 2
         self.expected_hash = message[0:digest_nybble_count]
         self.public_seed = message[digest_nybble_count:]
-        self.get_secret_from_user(self.addr[0], self.check_secret_from_user,
+        self.factory.get_secret_from_user(self.transport.getPeer().host,
+                self.check_secret_from_user,
                 self.send_secret_to_remote)
     
     def send_secret_to_remote(self, secret_message):
         """A callback for the invitation protocol to start a new phase 
         involving the other end getting the hash-digest of the public 
         seed and a secret we receive as a parameter."""
-
         h = hash()
         h.update(self.public_seed)
         h.update(secret_message)
-        self.send(h.hexdigest() + message_terminator)
+        self.sendLine(h.hexdigest())
         
     def check_secret_from_user(self, secret_message):
         """A callback for the invitation protocol to verify the secret
         that the user gives, against the hash we received over the
         network."""
-
         h = hash()
         h.update(secret_message)
         digest = h.hexdigest()
@@ -163,89 +115,111 @@
             h = hash()
             h.update(self.public_seed)
             h.update(secret_message)
-            self.send(h.hexdigest() + message_terminator)
+            self.sendLine(h.hexdigest())
 
             self.logging.debug("User knew secret!")
+
+            self.transport.loseConnection()
             return True
 
         self.logging.info("User secret %r is wrong.", secret_message)
         return False
 
-#    def handle_error(self, *args):
-#        self.logging.error("%s", args)
-
-
-class SendInvitation(asyncore.dispatcher_with_send):
-    """Narrative "Bob".
-    """
-
-    def __init__(self, host, port, auth_complete_cb, secret_message,
-            public_seed, on_close, local_map=False):
-        self.logging = logging.getLogger("SendInvitation")
-
-        if local_map:
-            if not hasattr(local_map, "get"):
-                local_map = dict()
-            self.__map = local_map
-            asyncore.dispatcher_with_send.__init__(self, map=self.__map)
-        else:
-            self.__map = None
-            asyncore.dispatcher_with_send.__init__(self)
-
-        self.auth_complete_cb = auth_complete_cb
+
+class ProcessAnInvitationFactory(ServerFactory):
+    """Hold configuration values for all the connections, and fire off a
+    protocol to handle the data sent and received."""
+
+    protocol = ProcessAnInvitationProtocol
+
+    def __init__(self, get_secret_from_user, on_close):
+        self.logging = logging.getLogger(self.__class__.__name__)
+        self.get_secret_from_user = get_secret_from_user
         self.on_close = on_close
-        
-        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.connect((host, port))
-
-        self.recv_buffers = []
-
-        self.logging.debug("secret is %r", secret_message)
-
-        h = hash()
-        h.update(secret_message)
-        self.send(h.hexdigest() + public_seed + message_terminator)
-
-        h = hash()
-        h.update(public_seed)
-        h.update(secret_message)
+
+
+class SendInvitationProtocol(basic.LineReceiver):
+    """Narrative "Bob"."""
+
+    def __init__(self):
+        """Initialize."""
+        self.logging = logging.getLogger(self.__class__.__name__)
+        self.logging.debug("initialized")
+
+    def connectionMade(self):
+        """Fire when a connection is made to the listener.  No obligation
+        here."""
+        self.logging.debug("connection made")
+
+        h = hash()
+        h.update(self.factory.secret_message)
+        self.sendLine(h.hexdigest() + self.factory.public_seed)
+
+        h = hash()
+        h.update(self.factory.public_seed)
+        h.update(self.factory.secret_message)
         self.expected_hash_of_secret = h.hexdigest()
 
-    def process(self):
-        """How we jump into asyncore processing.  Call this occasionally."""
-        asyncore.loop(0.01, False, self.__map, 3)
-        return True
-
-    def handle_connect(self):
-        self.logging.debug("connecting")
-
-    def handle_read(self):
-        message = receive_up_to_char(self, self.recv_buffers)
-        if message is None:
-            return
-        self.logging.debug("received %r", message)
+
+    def lineReceived(self, message):
+        """Handler for receipt of a message from the Alice end."""
         if message == self.expected_hash_of_secret:
-            remote_host = self.socket.getpeername()[0]
+            remote_host = self.transport.getPeer().host
             try:
                 remote_hostname = get_remote_hostname(remote_host)
             except dbus.exceptions.DBusException:
                 remote_hostname = None
-            self.auth_complete_cb(remote_hostname, "(port)", "(info)")
+            self.factory.auth_complete_cb(remote_hostname, "(port)", "(info)")
+            self.transport.loseConnection()
         else:
             self.logging.warn("Expected %r from invitation.",
                     self.expected_hash_of_secret)
 
-    def handle_close(self):
-        self.logging.debug("closing")
-        self.on_close()
-        self.close()
-
-#    def handle_error(self, *args):
-#        self.logging.error("%s", args)
-
-
-
-if __name__ == "__main__":
-    logging.basicConfig(level=logging.DEBUG, format=
-            "%(asctime)s [%(process)d] %(levelname)s - %(name)s %(message)s")  
-
+    def connectionLost(self):
+        """When a connected socked is broken, this is fired."""
+        self.logging.info("connection lost.")
+        basic.LineReceiver.connectionLost(self)
+
+
+class SendInvitationFactory(ReconnectingClientFactory):
+    """Hold configuration values for all the connections, and fire off a
+    protocol to handle the data sent and received."""
+
+    protocol = SendInvitationProtocol
+
+    def __init__(self, auth_complete_cb, secret_message, public_seed,
+            on_close):
+        self.logging = logging.getLogger(self.__class__.__name__)
+        self.auth_complete_cb = auth_complete_cb
+        self.secret_message = secret_message
+        self.public_seed = public_seed
+        self.on_close = on_close
+        self.logging.debug("initialized")
+
+    def close(self):
+        """Called from the UI when a window is destroyed and we do not need
+        this connection any more."""
+        self.logging.warn("close not handled properly")  # FIXME
+
+    def clientConnectionFailed(self, connector, reason):
+        """When we fail to connect to the listener, this is fired."""
+        self.logging.warn("connect failed. %s", reason)
+        ReconnectingClientFactory.clientConnectionFailed(self, connector,
+                reason)
+
+    def clientConnectionLost(self, connector, reason):
+        """When a connected socked is broken, this is fired."""
+        self.logging.info("connection lost. %s", reason)
+        ReconnectingClientFactory.clientConnectionLost(self, connector, reason)
+
+        
+def start_send_invitation(host, port, auth_complete_cb, secret_message,
+        public_seed, on_close):
+    """Instantiate the factory to hold configuration data about sending an
+    invitation and let the reactor add it to its event-handling loop by way of
+    starting a TCP connection."""
+    factory = SendInvitationFactory(auth_complete_cb, secret_message,
+            public_seed, on_close)
+    reactor.connectTCP(host, port, factory)
+
+    return factory

=== modified file 'desktopcouch/pair/pair.py'
--- desktopcouch/pair/pair.py	2009-07-08 17:48:11 +0000
+++ desktopcouch/pair/pair.py	2009-07-10 22:13:01 +0000
@@ -52,6 +52,11 @@
 import random
 import cgi
 
+from twisted.internet import gtk2reactor
+gtk2reactor.install()
+from twisted.internet.reactor import run as run_program
+from twisted.internet.reactor import stop as stop_program
+
 import pygtk
 pygtk.require('2.0')
 import gtk
@@ -64,7 +69,6 @@
 discovery_tool_version = "1"
 
 
-
 def generate_secret(length=7):
     """Create a secret that is easy to write and read.  We hate ambiguity and
     errors."""
@@ -124,7 +128,6 @@
 
     def destroy(self, widget, data=None):
         """The window is destroyed."""
-        gobject.source_remove(self.inviter_loop)
         self.inviter.close()
 
     def auth_completed(self, remote_host, remote_port, remote_info):
@@ -154,10 +157,9 @@
         self.secret_message = secret_message
         self.public_seed = generate_secret()
 
-        self.inviter = network_io.SendInvitation(hostname, port, 
+        self.inviter = network_io.start_send_invitation(hostname, port, 
                 self.auth_completed, self.secret_message, self.public_seed,
-                self.on_close, local_map=False)
-        self.inviter_loop = gobject.idle_add(self.inviter.process)
+                self.on_close)
 
         top_vbox = gtk.VBox()
         self.window.add(top_vbox)
@@ -173,6 +175,12 @@
         top_vbox.pack_start(text, False, False, 10)
         text.show()
 
+        cancel_button = gtk.Button(stock=gtk.STOCK_CANCEL)
+        cancel_button.set_border_width(3)
+        cancel_button.connect("clicked", lambda widget: self.window.destroy())
+        top_vbox.pack_end(cancel_button, False, False, 10)
+        cancel_button.show()
+
         self.window.show_all()
 
 
@@ -293,8 +301,6 @@
         if self.advertisement is not None:
             self.advertisement.die()
         self.listener.close()
-        if self.listener_loop is not None:
-            gobject.source_remove(self.listener_loop)
 
     def receive_invitation_challenge(self, remote_address, is_secret_valid,
             send_secret):
@@ -317,9 +323,7 @@
 
         self.listener = network_io.ListenForInvitations(
                 self.receive_invitation_challenge, 
-                lambda: self.window.destroy(),
-                local_map=False)
-        self.listener_loop = gobject.idle_add(self.listener.process)
+                lambda: self.window.destroy())
 
         listen_port = self.listener.get_local_port()
 
@@ -433,7 +437,7 @@
 
     def destroy(self, widget, data=None):
         """The window was destroyed."""
-        gtk.main_quit()
+        stop_program()
 
     def create_pick_pane(self, container):
         """Set up the pane that contains what's necessary to choose an
@@ -633,6 +637,7 @@
 def main(args):
     """Start execution."""
     global pick_or_listen  # pylint: disable-msg=W0601
+
     logging.basicConfig(level=logging.DEBUG, format=
             "%(asctime)s [%(process)d] %(name)s:%(levelname)s:  %(message)s")
 
@@ -641,7 +646,7 @@
     try:
         logging.debug("starting couchdb pairing tool")
         pick_or_listen = PickOrListen()
-        return gtk.main()
+        return run_program()
     finally:
         logging.debug("exiting couchdb pairing tool")
 

=== modified file 'desktopcouch/pair/tests/test_network_io.py'
--- desktopcouch/pair/tests/test_network_io.py	2009-07-08 17:48:11 +0000
+++ desktopcouch/pair/tests/test_network_io.py	2009-07-10 22:13:01 +0000
@@ -14,15 +14,30 @@
 # You should have received a copy of the GNU Lesser General Public License
 # along with desktopcouch.  If not, see <http://www.gnu.org/licenses/>.
 
+
+import pygtk
+pygtk.require('2.0')
+import gtk
+
+from twisted.internet import gtk2reactor
+gtk2reactor.install()
+from twisted.internet import reactor, task
+
+
 if __name__ == "__main__":
     import sys, os
     sys.path.append(os.pardir)
-    from couchdb_pairing.network_io import SendInvitation, ListenForInvitations
+    from couchdb_pairing.network_io import start_send_invitation, ListenForInvitations
+    from couchdb_pairing.dbus_io import get_local_hostname
 else:
-    from ..couchdb_pairing.network_io import SendInvitation, ListenForInvitations
+    from ..couchdb_pairing.network_io import start_send_invitation, ListenForInvitations
+    from ..couchdb_pairing.dbus_io import get_local_hostname
 
 import unittest
 
+local_hostname = ".".join(get_local_hostname())
+
+
 class TestNetworkIO(unittest.TestCase):
 
     def setUp(self):
@@ -47,11 +62,9 @@
                 listener_complete_auth()
 
         def listener_close_socket():
-            print "LISTNER!\n\n"
             self._listener_socket_state = "closed"
 
         def inviter_close_socket():
-            print "INVITER!\n\n"
             self._inviter_socket_state = "closed"
 
         def inviter_complete_auth(a, b, c):
@@ -67,16 +80,21 @@
 
         def inviter_display_message(*args):
             """Show message to user."""
-            print "display message from inviter:", args
+            logging.info("display message from inviter: %s", args)
 
-        self.inviter = SendInvitation("localhost", listener_port,
+        self.inviter = start_send_invitation(local_hostname, listener_port,
                 inviter_complete_auth, secret, "seed", inviter_close_socket)
 
-        for i in xrange(50):
+        def exit_on_success():
             if self._listener_auth_completed and self._inviter_auth_completed:
-                break
-            self.listener.process()
-            self.inviter.process()
+                reactor.stop()
+        task.LoopingCall(exit_on_success).start(1.0)
+
+        def exit_on_timeout():
+            reactor.stop()
+        reactor.callLater(30, exit_on_timeout)
+
+        reactor.run()
 
         self.assertTrue(self._listener_auth_completed)
         self.assertTrue(self._inviter_auth_completed)
@@ -117,14 +135,12 @@
 
         def inviter_display_message(*args):
             """Show message to user."""
-            print "display message from inviter:", args
+            logging.info("display message from inviter: %s", args)
 
-        self.inviter = SendInvitation("localhost", listener_port,
+        self.inviter = start_send_invitation(local_hostname, listener_port,
                 inviter_complete_auth, secret, "seed", inviter_close_socket)
 
-        for i in xrange(50):
-            self.listener.process()
-            self.inviter.process()
+        # FIXME
 
         self.assertFalse(self._listener_auth_completed)
         self.assertFalse(self._inviter_auth_completed)
@@ -147,11 +163,9 @@
 
 
         def listener_close_socket():
-            print "closed listener"
             self._listener_socket_state = "closed"
 
         def inviter_close_socket():
-            print "closed inviter"
             self._inviter_socket_state = "closed"
 
         def inviter_complete_auth(a, b, c):
@@ -167,16 +181,12 @@
 
         def inviter_display_message(*args):
             """Show message to user."""
-            print "display message from inviter:", args
+            logging.info("display message from inviter: %s", args)
 
-        self.inviter = SendInvitation("localhost", listener_port,
+        self.inviter = start_send_invitation(local_hostname, listener_port,
                 inviter_complete_auth, secret, "seed", inviter_close_socket)
 
-        for i in xrange(500):
-            if self._inviter_socket_state == "closed" and self._listener_socket_state == "closed":
-                break
-            self.listener.process()
-            self.inviter.process()
+        # FIXME
 
         self.assertFalse(self._listener_auth_completed)
         self.assertFalse(self._inviter_auth_completed)
@@ -185,8 +195,6 @@
         self.assertEqual(self._inviter_socket_state, "closed")
 
 
-
-
 if __name__ == "__main__":
     import logging
     logging.basicConfig(level=logging.ERROR)