separate blockchain and network
[electrum-nvc.git] / lib / interface.py
index c843679..faf165c 100644 (file)
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 
-import random, socket, ast, re, ssl
+import random, socket, ast, re, ssl, errno
 import threading, traceback, sys, time, json, Queue
 
-from version import ELECTRUM_VERSION
-from util import print_error
+from version import ELECTRUM_VERSION, PROTOCOL_VERSION
+from util import print_error, print_msg
 
 
 DEFAULT_TIMEOUT = 5
-DEFAULT_SERVERS = [ 
-    'electrum.novit.ro:50001:t', 
-    'electrum.pdmc.net:50001:t',
-    'ecdsa.org:50001:t',
-    'electrum.bitcoins.sk:50001:t',
-    'uncle-enzo.info:50001:t',
-    'electrum.bytesized-hosting.com:50001:t',
-    'electrum.bitcoin.cz:50001:t',
-    'electrum.bitfoo.org:50001:t'
-    ]
+DEFAULT_PORTS = {'t':'50001', 's':'50002', 'h':'8081', 'g':'8082'}
+
+DEFAULT_SERVERS = {
+    'the9ull.homelinux.org': {'h': '8082', 't': '50001'},
+    'electrum.coinwallet.me': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.dynaloop.net': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.koh.ms': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.novit.ro': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.stepkrav.pw': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'ecdsa.org': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.mooo.com': {'h': '8081', 't': '50001'},
+    'electrum.bitcoins.sk': {'h': '8081', 's': '50002', 't': '50001', 'g': '8'},
+    'electrum.no-ip.org': {'h': '80', 's': '50002', 't': '50001', 'g': '443'},
+    'electrum.drollette.com': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'btc.it-zone.org': {'h': '80', 's': '110', 't': '50001', 'g': '443'},
+    'electrum.yacoin.com': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'},
+    'electrum.be': {'h': '8081', 's': '50002', 't': '50001', 'g': '8082'}
+}
+
+
+
+def filter_protocol(servers, p):
+    l = []
+    for k, protocols in servers.items():
+        if p in protocols:
+            l.append( ':'.join([k, protocols[p], p]) )
+    return l
+    
+
 
 proxy_modes = ['socks4', 'socks5', 'http']
 
 
 def pick_random_server():
-    return random.choice( DEFAULT_SERVERS )
+    return random.choice( filter_protocol(DEFAULT_SERVERS,'s') )
 
 
 
 
 class Interface(threading.Thread):
 
+
     def register_callback(self, event, callback):
         with self.lock:
             if not self.callbacks.get(event):
                 self.callbacks[event] = []
             self.callbacks[event].append(callback)
 
+
     def trigger_callback(self, event):
         with self.lock:
             callbacks = self.callbacks.get(event,[])[:]
         if callbacks:
             [callback() for callback in callbacks]
 
+
     def init_server(self, host, port, proxy=None, use_ssl=True):
         self.host = host
         self.port = port
@@ -69,25 +91,77 @@ class Interface(threading.Thread):
         #json
         self.message_id = 0
         self.unanswered_requests = {}
+        #banner
+        self.banner = ''
+        self.pending_transactions_for_notifications= []
+
+
+    def parse_servers(self, result):
+        """ parse servers list into dict format"""
+
+        servers = {}
+        for item in result:
+            host = item[1]
+            out = {}
+            version = None
+            pruning_level = '-'
+            if len(item) > 2:
+                for v in item[2]:
+                    if re.match("[stgh]\d*", v):
+                        protocol, port = v[0], v[1:]
+                        if port == '': port = DEFAULT_PORTS[protocol]
+                        out[protocol] = port
+                    elif re.match("v(.?)+", v):
+                        version = v[1:]
+                    elif re.match("p\d*", v):
+                        pruning_level = v[1:]
+                    if pruning_level == '': pruning_level = '0'
+            try: 
+                is_recent = float(version)>=float(PROTOCOL_VERSION)
+            except:
+                is_recent = False
+
+            if out and is_recent:
+                out['pruning'] = pruning_level
+                servers[host] = out
 
+        return servers
 
 
     def queue_json_response(self, c):
 
         # uncomment to debug
-        # print "<--",c
+        # print_error( "<--",c )
 
         msg_id = c.get('id')
         error = c.get('error')
         
         if error:
             print_error("received error:", c)
+            if msg_id is not None:
+                with self.lock: 
+                    method, params, channel = self.unanswered_requests.pop(msg_id)
+                response_queue = self.responses[channel]
+                response_queue.put({'method':method, 'params':params, 'error':error, 'id':msg_id})
+
             return
 
         if msg_id is not None:
             with self.lock: 
                 method, params, channel = self.unanswered_requests.pop(msg_id)
             result = c.get('result')
+
+            if method == 'server.version':
+                self.server_version = result
+
+            elif method == 'server.banner':
+                self.banner = result
+                self.trigger_callback('banner')
+
+            elif method == 'server.peers.subscribe':
+                self.servers = self.parse_servers(result)
+                self.trigger_callback('peers')
+
         else:
             # notification: find the channel(s)
             method = c.get('method')
@@ -117,16 +191,19 @@ class Interface(threading.Thread):
                     return
                 
         response_queue = self.responses[channel]
-        response_queue.put({'method':method, 'params':params, 'result':result, 'id':msg_id})
+        response_queue.put((self, {'method':method, 'params':params, 'result':result, 'id':msg_id}))
 
 
 
     def get_response(self, channel='default', block=True, timeout=10000000000):
-        return self.responses[channel].get(block, timeout)
+        i, r = self.responses[channel].get(block, timeout)
+        return r
 
-    def register_channel(self, channel):
+    def register_channel(self, channel, queue=None):
+        if queue is None:
+            queue = Queue.Queue()
         with self.lock:
-            self.responses[channel] = Queue.Queue()
+            self.responses[channel] = queue
 
     def poke(self, channel):
         self.responses[channel].put(None)
@@ -135,8 +212,20 @@ class Interface(threading.Thread):
     def init_http(self, host, port, proxy=None, use_ssl=True):
         self.init_server(host, port, proxy, use_ssl)
         self.session_id = None
-        self.connection_msg = ('https' if self.use_ssl else 'http') + '://%s:%d'%( self.host, self.port )
         self.is_connected = True
+        self.connection_msg = ('https' if self.use_ssl else 'http') + '://%s:%d'%( self.host, self.port )
+        try:
+            self.poll()
+        except:
+            print_error("http init session failed")
+            self.is_connected = False
+            return
+
+        if self.session_id:
+            print_error('http session:',self.session_id)
+            self.is_connected = True
+        else:
+            self.is_connected = False
 
     def run_http(self):
         self.is_connected = True
@@ -162,10 +251,11 @@ class Interface(threading.Thread):
 
     def send_http(self, messages, channel='default'):
         import urllib2, json, time, cookielib
+        print_error( "send_http", messages )
         
         if self.proxy:
             import socks
-            socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
+            socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]) + 1, self.proxy["host"], int(self.proxy["port"]) )
             socks.wrapmodule(urllib2)
 
         cj = cookielib.CookieJar()
@@ -193,8 +283,11 @@ class Interface(threading.Thread):
         if self.session_id:
             headers['cookie'] = 'SESSION=%s'%self.session_id
 
-        req = urllib2.Request(self.connection_msg, data_json, headers)
-        response_stream = urllib2.urlopen(req)
+        try:
+            req = urllib2.Request(self.connection_msg, data_json, headers)
+            response_stream = urllib2.urlopen(req, timeout=DEFAULT_TIMEOUT)
+        except:
+            return
 
         for index, cookie in enumerate(cj):
             if cookie.name=='SESSION':
@@ -234,29 +327,32 @@ class Interface(threading.Thread):
             self.connection_msg += " using proxy %s:%s:%s"%(self.proxy.get('mode'), self.proxy.get('host'), self.proxy.get('port'))
             import socks
             s = socks.socksocket()
-            s.setproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
+            s.setproxy(proxy_modes.index(self.proxy["mode"]) + 1, self.proxy["host"], int(self.proxy["port"]) )
 
         if self.use_ssl:
-            s = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_SSLv23)
+            s = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_SSLv23, do_handshake_on_connect=True)
             
         s.settimeout(2)
         s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 
         try:
             s.connect(( self.host.encode('ascii'), int(self.port)))
-            s.settimeout(60)
-            self.s = s
-            self.is_connected = True
         except:
+            #traceback.print_exc(file=sys.stdout)
+            print_error("failed to connect", host, port)
             self.is_connected = False
             self.s = None
+            return
 
+        s.settimeout(60)
+        self.s = s
+        self.is_connected = True
 
     def run_tcp(self):
         try:
+            #if self.use_ssl: self.s.do_handshake()
             out = ''
             while self.is_connected:
-
                 try: 
                     timeout = False
                     msg = self.s.recv(1024)
@@ -264,10 +360,18 @@ class Interface(threading.Thread):
                     timeout = True
                 except ssl.SSLError:
                     timeout = True
+                except socket.error, err:
+                    if err.errno in [11, 10035]:
+                        print_error("socket errno", err.errno)
+                        time.sleep(0.1)
+                        continue
+                    else:
+                        traceback.print_exc(file=sys.stdout)
+                        raise
 
                 if timeout:
                     # ping the server with server.version, as a real ping does not exist yet
-                    self.send([('server.version', [ELECTRUM_VERSION])])
+                    self.send([('server.version', [ELECTRUM_VERSION, PROTOCOL_VERSION])])
                     continue
 
                 out += msg
@@ -299,22 +403,30 @@ class Interface(threading.Thread):
             self.unanswered_requests[self.message_id] = method, params, channel
             ids.append(self.message_id)
             # uncomment to debug
-            # print "-->",request
+            # print "-->", request
             self.message_id += 1
             out += request + '\n'
         while out:
             try:
                 sent = self.s.send( out )
                 out = out[sent:]
-            except:
-                # this happens when we get disconnected
-                print_error( "Not connected, cannot send" )
-                return None
+            except socket.error,e:
+                if e[0] in (errno.EWOULDBLOCK,errno.EAGAIN):
+                    print_error( "EAGAIN: retrying")
+                    time.sleep(0.1)
+                    continue
+                else:
+                    traceback.print_exc(file=sys.stdout)
+                    # this happens when we get disconnected
+                    print_error( "Not connected, cannot send" )
+                    return None
         return ids
 
 
 
-    def __init__(self, config=None, loop=False):
+    def __init__(self, config=None):
+        self.server = random.choice(filter_protocol(DEFAULT_SERVERS, 's'))
+        self.proxy = None
 
         if config is None:
             from simple_config import SimpleConfig
@@ -322,7 +434,6 @@ class Interface(threading.Thread):
 
         threading.Thread.__init__(self)
         self.daemon = True
-        self.loop = loop
         self.config = config
         self.connect_event = threading.Event()
 
@@ -333,9 +444,14 @@ class Interface(threading.Thread):
         self.callbacks = {}
         self.lock = threading.Lock()
 
-        self.servers = [] # actual list from IRC
+        self.servers = {} # actual list from IRC
         self.rtime = 0
         self.bytes_received = 0
+        self.is_connected = False
+
+        # init with None server, in case we are offline 
+        self.init_server(None, None)
+
 
 
 
@@ -343,30 +459,14 @@ class Interface(threading.Thread):
         if self.config.get('server'):
             self.init_with_server(self.config)
         else:
-            print "Using random server..."
-            servers = DEFAULT_SERVERS[:]
-            while servers:
-                server = random.choice( servers )
-                servers.remove(server)
-                self.config.set_key('server', server, False)
-                self.init_with_server(self.config)
-                if self.is_connected: break
-
-            if not servers:
-                print 'no server available'
-                self.is_connected = False
-                self.connect_event.set() # to finish start
-                self.server = 'ecdsa.org:50001:t'
-                self.proxy = None
-                return
+            if self.config.get('auto_cycle') is None:
+                self.config.set_key('auto_cycle', True, False)
+
+        if not self.is_connected: 
+            self.connect_event.set()
+            return
 
         self.connect_event.set()
-        if self.is_connected:
-            self.send([('server.version', [ELECTRUM_VERSION])])
-            self.trigger_callback('connected')
-        else:
-            self.trigger_callback('notconnected')
-            #print_error("Failed to connect " + self.connection_msg)
 
 
     def init_with_server(self, config):
@@ -400,7 +500,12 @@ class Interface(threading.Thread):
             with self.lock:
                 if self.subscriptions.get(channel) is None: 
                     self.subscriptions[channel] = []
-                self.subscriptions[channel] += sub
+                for message in sub:
+                    if message not in self.subscriptions[channel]:
+                        self.subscriptions[channel].append(message)
+
+        if not self.is_connected: 
+            return
 
         if self.protocol in 'st':
             with self.lock:
@@ -411,12 +516,6 @@ class Interface(threading.Thread):
 
         return out
 
-    def resend_subscriptions(self):
-        for channel, messages in self.subscriptions.items():
-            if messages:
-                self.send(messages, channel)
-
-
 
     def parse_proxy_options(self, s):
         if type(s) == type({}): return s  # fixme: type should be fixed
@@ -448,12 +547,25 @@ class Interface(threading.Thread):
             print "changing server:", server, proxy
             self.server = server
             self.proxy = proxy
-            if self.protocol in 'st':
+            if self.is_connected and self.protocol in 'st' and self.s:
                 self.s.shutdown(socket.SHUT_RDWR)
                 self.s.close()
             self.is_connected = False  # this exits the polling loop
             self.trigger_callback('disconnecting') # for actively disconnecting
 
+    def stop(self):
+        if self.is_connected and self.protocol in 'st' and self.s:
+            self.s.shutdown(socket.SHUT_RDWR)
+            self.s.close()
+
+
+    def get_servers(self):
+        if not self.servers:
+            return DEFAULT_SERVERS
+        else:
+            return self.servers
+
+
     def is_empty(self, channel):
         q = self.responses.get(channel)
         if q: 
@@ -491,22 +603,24 @@ class Interface(threading.Thread):
         return out
 
 
-    def start(self):
+    def start(self, queue):
+        self.queue = queue
         threading.Thread.start(self)
-        # wait until connection is established
-        self.connect_event.wait()
+
+
 
     def run(self):
-        while True:
-            self.init_interface()
-            if self.is_connected:
-                self.resend_subscriptions()
-                self.run_tcp() if self.protocol in 'st' else self.run_http()
+        self.init_interface()
+        if self.is_connected:
+            self.send([('server.version', [ELECTRUM_VERSION, PROTOCOL_VERSION])])
+            self.change_status()
+            self.run_tcp() if self.protocol in 'st' else self.run_http()
+        self.change_status()
+        
+    def change_status(self):
+        self.queue.put(self)
 
-            self.trigger_callback('disconnected')
 
-            if not self.loop: break
-            time.sleep(5)