Ticket #1969: diverse-connection.patch
| File diverse-connection.patch, 12.8 kB (added by sgala, 10 months ago) |
|---|
-
src/common/xmpp/transports_nb.py
old new 253 253 self.printed_error = False 254 254 255 255 # 0 - not connected 256 # 1 - connected 256 # 1 - connecting 257 # 2 - connected 257 258 # -1 - about to disconnect (when we wait for final events to complete) 258 259 # -2 - disconnected 259 260 self.state = 0 … … 274 275 # This prevents replug of same object with the same flags 275 276 self.writable = True 276 277 self.readable = False 278 self.ais = None 277 279 278 280 def plugin(self, owner): 279 281 ''' Fire up connection. Return non-empty string on success. … … 307 309 self.state = 0 308 310 success = False 309 311 try: 310 for ai in socket.getaddrinfo(server[0],server[1],socket.AF_UNSPEC,socket.SOCK_STREAM): 311 try: 312 self._sock=socket.socket(*ai[:3]) 313 self._sock.setblocking(False) 314 self._server=ai[4] 315 success = True 316 break 317 except: 318 if sys.exc_value[0] == errno.EINPROGRESS: 319 success = True 320 break 321 #for all errors, we try other addresses 322 continue 312 self.set_timeout(CONNECT_TIMEOUT_SECONDS) 313 if len(server) == 2 and type(server[0]) in (str, unicode) and not self.ais: 314 # FIXME: blocks here 315 print "blocking getaddrinfo", server 316 self.ais = socket.getaddrinfo(server[0],server[1],socket.AF_UNSPEC,socket.SOCK_STREAM) 317 #print "self.ais=", self.ais 318 else: 319 self.ais = (server,) 320 return self._do_connect() 323 321 except socket.gaierror, e: 324 322 log.info("Lookup failure for %s: %s[%s]", self.getName(), e[1], repr(e[0]), exc_info=True) 325 323 except: … … 328 326 if not success: 329 327 if self.on_connect_failure: 330 328 self.on_connect_failure() 331 return False 332 333 self.fd = self._sock.fileno() 334 self.idlequeue.plug_idle(self, True, False) 335 self.set_timeout(CONNECT_TIMEOUT_SECONDS) 336 self._do_connect() 337 return True 329 return success 338 330 339 331 def _plug_idle(self): 340 332 readable = self.state != 0 … … 540 532 541 533 def _do_connect(self): 542 534 if self.state != 0: 535 print "do_connect while connected" 543 536 return 544 self._sock.setblocking(False) 545 self._send = self._sock.send 546 self._recv = self._sock.recv 547 errnum = 0 548 try: 549 self._sock.connect(self._server) 550 except socket.error, e: 551 errnum = e[0] 537 for ai in self.ais: 538 success = False 539 try: 540 self._sock=socket.socket(*ai[:3]) 541 self._sock.setblocking(False) 542 self._server=ai[4] 543 self.fd = self._sock.fileno() 544 self.idlequeue.plug_idle(self, True, False) 545 self._send = self._sock.send 546 self._recv = self._sock.recv 547 errnum = 0 548 549 self._sock.connect(self._server) 550 print "connected" 551 success = True 552 self.state = 1 553 break 554 except socket.error, e: 555 #print "exception1",sys.exc_value 556 errnum = e[0] 552 557 553 # Ignore "Socket already connected". 554 # FIXME: This happens when we switch an already 555 # connected socket to SSL (STARTTLS). Instead of 556 # ignoring the error, the socket should only be 557 # connected to once. See #2846 and #3396. 558 workaround = (errno.EALREADY, 10056, 56) 559 560 # 10035 - winsock equivalent of EINPROGRESS 561 if errnum not in (errno.EINPROGRESS, 10035) + workaround: 562 log.error("_do_connect:", exc_info=True) 563 #traceback.print_exc() 564 # in progress, or would block 565 if errnum in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK): 566 return 567 # 10056 - already connected, only on win32 568 # code 'WS*' is not available on GNU, so we use its numeric value 569 elif errnum not in (0, 10056, errno.EISCONN): 570 self.remove_timeout() 558 # Ignore "Socket already connected". 559 # FIXME: This happens when we switch an already 560 # connected socket to SSL (STARTTLS). Instead of 561 # ignoring the error, the socket should only be 562 # connected to once. See #2846 and #3396. 563 workaround = (errno.EALREADY, 10056, 56) 564 565 # 10035 - winsock equivalent of EINPROGRESS 566 if errnum not in (errno.EINPROGRESS, 10035) + workaround: 567 log.error("_do_connect:", exc_info=True) 568 #traceback.print_exc() 569 # in progress, or would block 570 if errnum in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK): 571 success = True 572 self.state = 1 573 break 574 # 10056 - already connected, only on win32 575 # code 'WS*' is not available on GNU, so we use its numeric value 576 elif errnum not in (0, 10056, errno.EISCONN): 577 continue 578 579 self.remove_timeout() 580 if not success: 571 581 if self.on_connect_failure: 572 582 self.on_connect_failure() 573 583 return 574 self.remove_timeout()575 584 self._owner.Connection=self 576 585 self.state = 1 577 578 586 self._sock.setblocking(False) 579 587 self._plug_idle() 580 588 if self.on_connect: 581 589 self.on_connect() 582 590 self.on_connect = None 583 591 return True 584 592 585 593 def send(self, raw_data, now = False): … … 741 749 log.debug("_startSSL_pyOpenSSL called") 742 750 tcpsock = self._owner.Connection 743 751 # FIXME: should method be configurable? 744 tcpsock._sslContext = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_METHOD)745 #tcpsock._sslContext = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD)752 #tcpsock._sslContext = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_METHOD) 753 tcpsock._sslContext = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD) 746 754 tcpsock.ssl_errnum = 0 747 755 tcpsock._sslContext.set_verify(OpenSSL.SSL.VERIFY_PEER, self._ssl_verify_callback) 748 756 cacerts = os.path.join(gajim.DATA_DIR, 'other', 'cacerts.pem') -
src/common/connection.py
old new 481 481 # SRV resolver 482 482 self._proxy = proxy 483 483 self._secure = secur 484 self._hosts = [ {'host': h, 'port': p, 'prio': 10, 'weight': 10} ]485 484 self._hostname = hostname 485 def _on_resolve(host, result_array): 486 # SRV query returned at least one valid result, we put it in hosts dict 487 if len(result_array) != 0: 488 self._hosts = [i for i in result_array] 489 else: 490 # TODO: Make async query for non SVR servers and protect exceptions in getaddrinfo 491 self._hosts = [ {'host': h, 'port': p, 'prio': 10, 'weight': 10, 'ai': ai} for ai in socket.getaddrinfo(h,p,socket.AF_UNSPEC,socket.SOCK_STREAM) ] 492 self.connect_to_next_host() 486 493 if use_srv: 487 494 # add request for srv query to the resolve, on result '_on_resolve' 488 495 # will be called 489 496 gajim.resolver.resolve('_xmpp-client._tcp.' + helpers.idn_to_ascii(h), 490 self._on_resolve)497 _on_resolve) 491 498 else: 492 self._on_resolve('', [])499 gajim.resolver.resolve(helpers.idn_to_ascii(h), _on_resolve) 493 500 494 def _on_resolve(self, host, result_array):495 # SRV query returned at least one valid result, we put it in hosts dict496 if len(result_array) != 0:497 self._hosts = [i for i in result_array]498 self.connect_to_next_host()499 501 500 502 def on_proxy_failure(self, reason): 501 503 log.debug('Connection to proxy failed') … … 512 514 self.last_connection.socket.disconnect() 513 515 self.last_connection = None 514 516 self.connection = None 517 debug = [] 515 518 if gajim.verbose: 516 519 con = common.xmpp.NonBlockingClient(self._hostname, caller = self, 517 520 on_connect = self.on_connect_success, … … 535 538 con.RegisterDisconnectHandler(self._on_new_account) 536 539 537 540 log.info("Connecting to %s: [%s:%d]", self.name, host['host'], host['port']) 538 con.connect((host['host'], host['port']), proxy = self._proxy, 539 secure = self._secure) 541 if 'ai' in host: 542 con.connect(host['ai'], proxy = self._proxy, 543 secure = self._secure) 544 else: 545 con.connect((host['host'],host['port']), proxy = self._proxy, 546 secure = self._secure) 540 547 else: 541 548 if not retry and self.retrycount == 0: 542 549 log.debug("Out of hosts, giving up connecting to %s", self.name) -
src/common/nslookup.py
old new 20 20 import sys 21 21 import os 22 22 import re 23 import socket 23 24 24 25 from xmpp.idlequeue import * 25 26 … … 96 97 prop_value = prop_value[:-1] 97 98 current_host['host'] = prop_value 98 99 if len(current_host) == 4: 99 hosts.append(current_host) 100 current_host = None 100 try: 101 for ai in socket.getaddrinfo(current_host['host'],current_host['port'],socket.AF_UNSPEC,socket.SOCK_STREAM): 102 hosts.append(current_host) 103 hosts[-1]['ai'] = ai 104 finally: 105 current_host = None 101 106 return hosts 102 107 103 108 def _parse_srv_result_posix(self, fqdn, result): … … 129 134 port = int(port) 130 135 except ValueError: 131 136 continue 132 hosts.append({'host': host, 'port': port,'weight': weight, 133 'prio': prio}) 137 try: 138 for ai in socket.getaddrinfo(host,port,socket.AF_UNSPEC,socket.SOCK_STREAM): 139 hosts.append({'host': host, 'port': port,'weight': weight, 140 'prio': prio, 'ai':ai}) 141 except: 142 print "GAI failed", host, port 134 143 return hosts 135 144 136 145 def _on_ready(self, host, result): … … 290 299 if self.result_handler: 291 300 self.result_handler(self.host, self.result) 292 301 self.result_handler = None 302 303 class GetAddrInfo(IdleCommand): 304 def __init__(self, on_result, host='_xmpp-client', port=5222, type = 'srv'): 305 IdleCommand.__init__(self, on_result) 306 self.commandtimeout = 10 307 self.host = host.lower() 308 self.port = port 309 self.type = type.lower() 310 if not host_pattern.match(self.host): 311 # invalid host name 312 print >> sys.stderr, 'Invalid host: %s' % self.host 313 self.canexecute = False 314 return 315 if not ns_type_pattern.match(self.type): 316 print >> sys.stderr, 'Invalid querytype: %s' % self.type 317 self.canexecute = False 318 return 319 def start(self): 320 if not self.canexecute: 321 self.result = None 322 self._return_result() 323 return 324 def do_work(): 325 print "in do_work" 326 try: 327 self.result = socket.getaddrinfo(self.host, 328 self.port, 329 socket.AF_UNSPEC, 330 socket.SOCK_STREAM) 331 except: 332 print "Error", sys.exc_info() 333 finally: 334 print "Ended" 335 print "after do_work" 336 self.end() 337 print "before start_new_thread" 338 import threading 339 t = threading.Thread(target=do_work) 340 print t 341 t.start() 342 print "the thread is running" 343 def pollin(self): 344 if self.result != None: 345 self.end() 346 def end(self): 347 #self.idlequeue.unplug_idle(self.fd) 348 print "in self.end", self.result 349 350 351 class AIResolver: 352 def __init__(self, idlequeue): 353 self.idlequeue = idlequeue 354 # dict {host : list of addrinfo records} 355 self.resolved_hosts = {} 356 # dict {host : list of callbacks} 357 self.handlers = {} 358 def start_resolve(self, host,port): 359 ''' spawn new nslookup process and start waiting for results ''' 360 ns = GetAddrInfo(self._on_ready, host,port) 361 ns.set_idlequeue(self.idlequeue) 362 ns.commandtimeout = 10 363 ns.start() 364 def resolve(self, host, port, on_ready): 365 if host == '': # None means localhost 366 # empty host, return empty list of srv records 367 on_ready([]) 368 return 369 if host in self.resolved_hosts: 370 # host is already resolved, return cached values 371 on_ready(host, self.resolved_hosts[host]) 372 return 373 if host in self.handlers: 374 # host is about to be resolved by another connection, 375 # attach our callback (FIXME: synced on self.handlers) 376 self.handlers[host].append(on_ready) 377 else: 378 # host has never been resolved, start now 379 self.handlers[host] = [on_ready] 380 self.start_resolve(host,port) 381 def _on_ready(self, host, result): 382 print "AIResolver:", host, result 383 # practically it is impossible to be the opposite, but who knows :) 384 if not host in self.resolved_hosts: 385 self.resolved_hosts[host] = result_list 386 if host in self.handlers: 387 for callback in self.handlers[host]: 388 callback(host, result_list) 389 del self.handlers[host] 390 391 392 293 393 294 394 # below lines is on how to use API and assist in testing 295 395 if __name__ == '__main__': … … 301 401 import gobject 302 402 import gtk 303 403 304 resolver = Resolver(idlequeue) 404 #resolver = Resolver(idlequeue) 405 resolver = AIResolver(idlequeue) 305 406 306 407 def clicked(widget): 307 408 global resolver 308 409 host = text_view.get_text() 410 port = 5222 309 411 def on_result(host, result_array): 310 412 print 'Result:\n' + repr(result_array) 311 resolver.resolve(host, on_result)413 resolver.resolve(host, port, on_result) 312 414 win = gtk.Window() 313 415 win.set_border_width(6) 314 416 text_view = gtk.Entry() … … 321 423 but.connect('clicked', clicked) 322 424 win.add(hbox) 323 425 win.show_all() 324 gobject.timeout_add(200, idlequeue.process) 426 win.connect('destroy', lambda w: gtk.main_quit()) 427 gobject.timeout_add(500, idlequeue.process) 325 428 gtk.main()
