handle disconnections

client/electrum

@@ -24,6 +24,7 @@ from decimal import Decimal
 
 
 from wallet import format_satoshis
+from interface import loop_interfaces_thread, new_interface
 
 if __name__ == '__main__':
     known_commands = ['help', 'validateaddress', 'balance', 'contacts', 'create', 'restore', 'payto', 'sendtx', 'password', 'addresses', 'history', 'label', 'mktx','seed','import','signmessage','verifymessage','eval']
@@ -46,8 +47,7 @@ if __name__ == '__main__':
     wallet.set_path(options.wallet_path)
     wallet.read()
     wallet.remote_url = options.remote_url
-
+    interface = wallet.interface = new_interface(wallet)
-    interface = wallet.interface
 
     if len(args)==0:
         url = None
@@ -60,6 +60,8 @@ if __name__ == '__main__':
         firstarg = args[1] if len(args) > 1 else ''
         
     if cmd == 'gui':
+        import thread
+
         if options.gui=='gtk':
             import gui
         elif options.gui=='qt':
@@ -68,7 +70,7 @@ if __name__ == '__main__':
             print "unknown gui", options.gui
             exit(1)
 
-        interface.get_servers()
+        thread.start_new_thread(loop_interfaces_thread, (wallet,))
         gui = gui.ElectrumGui(wallet)
 
         try:
@@ -83,7 +85,6 @@ if __name__ == '__main__':
 
         if not found: exit(1)
 
-        interface.start(wallet)
         gui.main(url)
         wallet.save()
         sys.exit(0)
@@ -163,7 +164,7 @@ if __name__ == '__main__':
 
     # open session
     if cmd not in ['password', 'mktx', 'history', 'label', 'contacts', 'help', 'validateaddress', 'signmessage', 'verifymessage', 'eval', 'create']:
-        interface.new_session(wallet.all_addresses(), wallet.electrum_version)
+        interface.start_session(wallet)
         interface.update_wallet(wallet)
         wallet.save()
 

client/gui_qt.py

@@ -940,7 +940,6 @@ class ElectrumGui():
         wallet.save()
         return True
 
-
     def main(self,url):
         s = Timer()
         s.start()

client/interface.py

@@ -53,6 +53,9 @@ class Interface:
             self.port = port
             self.is_connected = False
 
+    def start_session(self, wallet):
+        pass
+
 
 class NativeInterface(Interface):
     """This is the original Electrum protocol. It uses polling, and a non-persistent tcp connection"""
@@ -62,10 +65,13 @@ class NativeInterface(Interface):
         if host: self.host = host
         self.port = port
 
-    def new_session(self, addresses, version):
+    def start_session(self, wallet):
+        addresses = wallet.all_addresses()
+        version = wallet.electrum_version
         self.is_up_to_date = False
         out = self.handler('session.new', [ version, addresses ] )
         self.session_id, self.message = ast.literal_eval( out )
+        self.update_wallet(wallet)
 
     def update_session(self, addresses):
         out = self.handler('session.update', [ self.session_id, addresses ] )
@@ -130,11 +136,11 @@ class NativeInterface(Interface):
         self.is_up_to_date = True
         return changed_addr
 
-    def update_wallet_thread(self, wallet):
+    def loop_sessions_thread(self, wallet):
         while True:
             try:
                 self.is_connected = False
-                self.new_session(wallet.all_addresses(), wallet.electrum_version)
+                self.start_session(wallet)
             except socket.error:
                 print "Not connected"
                 time.sleep(self.poll_interval())
@@ -167,9 +173,6 @@ class NativeInterface(Interface):
                     traceback.print_exc(file=sys.stdout)
                     break
                 
-    def start(self, wallet):
-        thread.start_new_thread(self.update_wallet_thread, (wallet,))
-
     def get_servers(self):
         thread.start_new_thread(self.update_servers_thread, ())
 
@@ -204,9 +207,16 @@ class TCPInterface(Interface):
         Interface.__init__(self)
         if host: self.host = host
         self.port = 50001
-        self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
-        self.s.connect(( self.host, self.port))
         self.tx_event = threading.Event()
+        self.disconnected_event = threading.Event()
+        self.disconnected_event.clear()
+        
+        self.addresses_waiting_for_status = []
+        self.addresses_waiting_for_history = []
+        # up to date
+        self.is_up_to_date = False
+        self.up_to_date_event = threading.Event()
+        self.up_to_date_event.clear()
 
     def send(self, cmd, params = []):
         request = json.dumps( { 'method':cmd, 'params':params } )
@@ -259,34 +269,48 @@ class TCPInterface(Interface):
                     elif cmd =='address.subscribe':
                         addr = c.get('address')
                         status = c.get('status')
+                        if addr in self.addresses_waiting_for_status:
+                            self.addresses_waiting_for_status.remove(addr)
                         if wallet.status.get(addr) != status:
-                            self.send('address.get_history', addr)
                             wallet.status[addr] = status
-                            self.is_up_to_date = False
+                            self.send('address.get_history', addr)
-                        else:
+                            self.addresses_waiting_for_history.append(addr) 
-                            self.is_up_to_date = True
 
                     elif cmd == 'address.get_history':
                         addr = c.get('address')
-                        print "updating history for", addr
+                        if addr in self.addresses_waiting_for_history:
-                        wallet.history[addr] = c.get('result')
+                            self.addresses_waiting_for_history.remove(addr)
+                        wallet.history[addr] = data
                         wallet.synchronize()
                         wallet.update_tx_history()
                         wallet.save()
                         self.was_updated = True
-                else:
+                    else:
-                    print "received message:", c
+                        print "received message:", c
+
+                    if self.addresses_waiting_for_status or self.addresses_waiting_for_history:
+                        self.is_up_to_date = False
+                    else:
+                        self.is_up_to_date = True
+                        self.up_to_date_event.set()
         except:
             traceback.print_exc(file=sys.stdout)
             self.is_connected = False
+            self.disconnected_event.set()
+
+    def update_wallet(self,wallet):
+        self.up_to_date_event.wait()
 
     def subscribe(self,address):
         self.send('address.subscribe', address)
+        self.addresses_waiting_for_status.append(address)
         
     def get_servers(self):
         self.send('server.peers')
 
-    def start(self, wallet):
+    def start_session(self, wallet):
+        self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
+        self.s.connect(( self.host, self.port))
         thread.start_new_thread(self.listen_thread, (wallet,))
         self.send('client.version', wallet.electrum_version)
         self.send('server.banner')
@@ -295,6 +319,8 @@ class TCPInterface(Interface):
             self.subscribe(address)
 
 
+
+
 class HttpInterface(Interface):
 
     def __init__(self):
@@ -316,3 +342,34 @@ class HttpInterface(Interface):
         self.rtime = time.time() - t1
         self.is_connected = True
         return out
+
+
+
+def new_interface(wallet):
+    host = wallet.host
+    port = wallet.port
+    if port == 50000:
+        interface = NativeInterface(host,port)
+    elif port == 50001:
+        interface = TCPInterface(host,port)
+    elif port in [80,8080,81,8181]:
+        interface = HttpInterface(host,port)            
+    else:
+        raise BaseException("unknown protocol: %d"%port)
+    return interface
+       
+
+def loop_interfaces_thread(wallet):
+
+    while True:
+        try:
+            wallet.interface.start_session(wallet)
+            wallet.interface.get_servers()
+        except socket.error:
+            print "Not connected"
+            time.sleep(5)
+            continue
+        wallet.interface.disconnected_event.wait()
+        print "Disconnected"
+        wallet.interface = new_interface(wallet)
+

client/wallet.py

@@ -543,7 +543,7 @@ class Wallet:
         f.close()
 
     def read(self):
-        from interface import NativeInterface, HttpInterface,TCPInterface
+        import interface
 
         upgrade_msg = """This wallet seed is deprecated. Please run upgrade.py for a diagnostic."""
         self.file_exists = False
@@ -552,7 +552,8 @@ class Wallet:
             data = f.read()
             f.close()
         except:
-            self.interface = NativeInterface()
+            #self.interface = NativeInterface()
+            self.port = 50000
             return
         try:
             d = ast.literal_eval( data )
@@ -561,8 +562,8 @@ class Wallet:
             self.use_encryption = d.get('use_encryption')
             self.fee = int( d.get('fee') )
             self.seed = d.get('seed')
-            host = d.get('host')
+            self.host = d.get('host')
-            port = d.get('port')
+            self.port = d.get('port')
             blocks = d.get('blocks')
             self.addresses = d.get('addresses')
             self.change_addresses = d.get('change_addresses')
@@ -585,15 +586,7 @@ class Wallet:
         if self.remote_url: assert self.master_public_key.encode('hex') == self.get_remote_mpk()
 
         self.file_exists = True
-
+        #self.interface = interface.start_interface(self)
-        if port == 50000:
-            self.interface = NativeInterface(host,port)
-        elif port == 50001:
-            self.interface = TCPInterface(host,port)
-        elif port in [80,8080,81,8181]:
-            self.interface = HttpInterface(host,port)            
-        else:
-            raise BaseException("unknown protocol: %d"%port)
 
         
     def get_new_address(self):