Changeset 9867

Timestamp:

06/30/08 02:02:32 (5 months ago)

Author:

tomk

Message:

- Refactored non-blocking transport and client classes - getaddrinfo is called
in Client now
- Added NonBlockingHttpBOSH transport (to tranports_nb) and BOSHClient
(to client_nb)
- Extended possible proxy types in configuration by "BOSH" proxy
- Rewrote NonBlockingTLS to invoke success callback only after successful TLS handshake is over (formerly, the TLS Plugin returned right after sending <starttls>)

Location:

branches/bosh_support

Files:

• data/glade/manage_proxies_window.glade (modified) (1 diff)
• src/common/connection.py (modified) (9 diffs)
• src/common/xmpp/auth_nb.py (modified) (1 diff)
• src/common/xmpp/client_nb.py (modified) (8 diffs)
• src/common/xmpp/debug.py (modified) (1 diff)
• src/common/xmpp/dispatcher_nb.py (modified) (8 diffs)
• src/common/xmpp/idlequeue.py (modified) (1 diff)
• src/common/xmpp/protocol.py (modified) (2 diffs)
• src/common/xmpp/transports_nb.py (modified) (15 diffs)
• src/common/xmpp/transports_new.py (deleted)
• src/config.py (modified) (2 diffs)
• test/test_client_nb.py (modified) (11 diffs)
• test/test_nonblockingtcp.py (modified) (5 diffs)

branches/bosh_support/data/glade/manage_proxies_window.glade

@@ -212 +212 @@
                               <property name="visible">True</property>
                               <property name="items" translatable="yes">HTTP Connect
-SOCKS5</property>
+SOCKS5
+BOSH</property>
                               <property name="add_tearoffs">False</property>
                               <property name="focus_on_click">True</property>

branches/bosh_support/src/common/connection.py

@@ -56 +56 @@
 import logging
 log = logging.getLogger('gajim.c.connection')
+log.setLevel(logging.DEBUG)
 
 ssl_error = {
@@ -208 +209 @@
         def _disconnectedReconnCB(self):
                 '''Called when we are disconnected'''
-                log.debug('disconnectedReconnCB')
+                log.error('disconnectedReconnCB')
                 if gajim.account_is_connected(self.name):
                         # we cannot change our status to offline or connecting
@@ -468 +469 @@
                 else:
                         proxy = None
-
                 h = hostname
                 p = 5222
@@ -505 +505 @@
 
         def on_proxy_failure(self, reason):
-                log.debug('Connection to proxy failed')
+                log.error('Connection to proxy failed: %s' % reason)
                 self.time_to_reconnect = None
                 self.on_connect_failure = None
@@ -520 +520 @@
                                 self.last_connection = None
                                 self.connection = None
-                        if gajim.verbose:
-                                con = common.xmpp.NonBlockingClient(self._hostname, caller = self,
-                                        on_connect = self.on_connect_success,
-                                        on_proxy_failure = self.on_proxy_failure,
-                                        on_connect_failure = self.connect_to_next_type)
-                        else:
-                                con = common.xmpp.NonBlockingClient(self._hostname, debug = [],
-                                        caller = self, on_connect = self.on_connect_success,
-                                        on_proxy_failure = self.on_proxy_failure,
-                                        on_connect_failure = self.connect_to_next_type)
-                        self.last_connection = con
-                        # increase default timeout for server responses
-                        common.xmpp.dispatcher_nb.DEFAULT_TIMEOUT_SECONDS = self.try_connecting_for_foo_secs
-                        con.set_idlequeue(gajim.idlequeue)
-                        # FIXME: this is a hack; need a better way
-                        if self.on_connect_success == self._on_new_account:
-                                con.RegisterDisconnectHandler(self._on_new_account)
 
                         if self._current_type == 'ssl':
@@ -547 +530 @@
                                 else:
                                         secur = None
+
+                        if self._proxy and self._proxy['type'] == 'bosh': 
+                                clientClass = common.xmpp.BOSHClient
+                        else:
+                                clientClass = common.xmpp.NonBlockingClient
+
+                        if gajim.verbose:
+                                con = common.xmpp.NonBlockingClient(
+                                        hostname=self._current_host['host'],
+                                        port=port,
+                                        caller=self,
+                                        idlequeue=gajim.idlequeue)
+                        else:
+                                con = common.xmpp.NonBlockingClient(
+                                        hostname=self._current_host['host'],
+                                        debug=[],
+                                        port=port,
+                                        caller=self,
+                                        idlequeue=gajim.idlequeue)
+
+                        self.last_connection = con
+                        # increase default timeout for server responses
+                        common.xmpp.dispatcher_nb.DEFAULT_TIMEOUT_SECONDS = self.try_connecting_for_foo_secs
+                        # FIXME: this is a hack; need a better way
+                        if self.on_connect_success == self._on_new_account:
+                                con.RegisterDisconnectHandler(self._on_new_account)
+                        
                         log.info('Connecting to %s: [%s:%d]', self.name,
                                 self._current_host['host'], port)
-                        con.connect((self._current_host['host'], port), proxy=self._proxy,
+                        con.connect(
+                                on_connect=self.on_connect_success,
+                                on_proxy_failure=self.on_proxy_failure,
+                                on_connect_failure=self.connect_to_next_type,
+                                proxy=self._proxy,
                                 secure = secur)
                 else:
@@ -562 +576 @@
                         else:
                                 self._connection_types = ['tls', 'ssl', 'plain']
+                        
+                        # FIXME: remove after tls and ssl will be degubbed
+                        #self._connection_types = ['plain']
                         host = self.select_next_host(self._hosts)
                         self._current_host = host
@@ -976 +993 @@
                                 self.remove_all_transfers()
                                 self.time_to_reconnect = None
-                                self.connection.start_disconnect(p, self._on_disconnected)
+
+                                self.connection.RegisterDisconnectHandler(self._on_disconnected)
+                                self.connection.send(p)
+                                self.connection.StreamTerminate()
+                                #self.connection.start_disconnect(p, self._on_disconnected)
                         else:
                                 self.time_to_reconnect = None
@@ -1011 +1032 @@
                 ''' called when a disconnect request has completed successfully'''
                 self.dispatch('STATUS', 'offline')
-                self.disconnect()
+                self.disconnect(on_purpose=True)
 
         def get_status(self):

branches/bosh_support/src/common/xmpp/auth_nb.py

@@ -170 +170 @@
                         self.DEBUG('Successfully authenticated with remote server.', 'ok')
                         handlers=self._owner.Dispatcher.dumpHandlers()
+                        print '6' * 79
+                        print handlers
+                        print '6' * 79
                         self._owner.Dispatcher.PlugOut()
                         dispatcher_nb.Dispatcher().PlugIn(self._owner)

branches/bosh_support/src/common/xmpp/client_nb.py

@@ -18 +18 @@
 
 '''
-Provides PlugIn class functionality to develop extentions for xmpppy.
-Also provides Client and Component classes implementations as the
-examples of xmpppy structures usage.
+Provides Client classes implementations as examples of xmpppy structures usage.
 These classes can be used for simple applications "AS IS" though.
 '''
@@ -26 +24 @@
 import socket
 import debug
-
-import transports_nb, dispatcher_nb, auth_nb, roster_nb
+import random
+
+import transports_nb, dispatcher_nb, auth_nb, roster_nb, protocol
 from client import *
+
+import logging
+log = logging.getLogger('gajim.c.x.client_nb')
+
+consoleloghandler = logging.StreamHandler()
+consoleloghandler.setLevel(logging.DEBUG)
+consoleloghandler.setFormatter(
+        logging.Formatter('%(levelname)s: %(message)s')
+)
+log.setLevel(logging.DEBUG)
+log.addHandler(consoleloghandler)
+log.propagate = False
+
 
 class NBCommonClient:
         ''' Base for Client and Component classes.'''
-        def __init__(self, server, port=5222, debug=['always', 'nodebuilder'], caller=None, 
-                on_connect=None, on_proxy_failure=None, on_connect_failure=None):
-                ''' Caches server name and (optionally) port to connect to. "debug" parameter specifies
-                        the debug IDs that will go into debug output. You can either specifiy an "include"
-                        or "exclude" list. The latter is done via adding "always" pseudo-ID to the list.
-                        Full list: ['nodebuilder', 'dispatcher', 'gen_auth', 'SASL_auth', 'bind', 'socket', 
-                         'CONNECTproxy', 'TLS', 'roster', 'browser', 'ibb'] . '''
-                
-                if isinstance(self, NonBlockingClient): 
-                        self.Namespace, self.DBG = 'jabber:client', DBG_CLIENT
-                elif isinstance(self, NBCommonClient): 
-                        self.Namespace, self.DBG = dispatcher_nb.NS_COMPONENT_ACCEPT, DBG_COMPONENT
-                
+        def __init__(self, hostname, idlequeue, port=5222, debug=['always', 'nodebuilder'], caller=None):
+                
+                ''' Caches connection data:
+                :param hostname: hostname of machine where the XMPP server is running (from Account
+                        of from SRV request) and port to connect to.
+                :param idlequeue: processing idlequeue
+                :param port: port of listening XMPP server
+                :param debug: specifies the debug IDs that will go into debug output. You can either
+                        specifiy an "include" or "exclude" list. The latter is done via adding "always" 
+                        pseudo-ID to the list. Full list: ['nodebuilder', 'dispatcher', 'gen_auth', 
+                        'SASL_auth', 'bind', 'socket', 'CONNECTproxy', 'TLS', 'roster', 'browser', 'ibb'].
+                        TODO: get rid of debug.py using
+                :param caller: calling object - it has to implement certain methods (necessary?)
+                        
+                '''
+                
+                self.DBG = DBG_CLIENT
+
+                self.Namespace = protocol.NS_CLIENT
+
+                self.idlequeue = idlequeue
                 self.defaultNamespace = self.Namespace
                 self.disconnect_handlers = []
-                self.Server = server
+
+                # XMPP server and port from account or SRV
+                self.Server = hostname
                 self.Port = port
                 
-                # Who initiated this client
-                # Used to register the EventDispatcher
+                # caller is who initiated this client, it is sed to register the EventDispatcher
                 self._caller = caller
                 if debug and type(debug) != list: 
@@ -63 +84 @@
                 self.connected = ''
                 self._component=0
-                self.idlequeue = None
                 self.socket = None
-                self.on_connect = on_connect
-                self.on_proxy_failure = on_proxy_failure
-                self.on_connect_failure = on_connect_failure
-                
-        def set_idlequeue(self, idlequeue):
-                self.idlequeue = idlequeue
+                self.on_connect = None
+                self.on_proxy_failure = None
+                self.on_connect_failure = None
+                self.proxy = None
+                
         
-        def disconnected(self):
-                ''' Called on disconnection. Calls disconnect handlers and cleans things up. '''
+        def on_disconnect(self):
+                '''
+                Called on disconnection - when connect failure occurs on running connection
+                (after stream is successfully opened).
+                Calls disconnect handlers and cleans things up.
+                '''
+                
                 self.connected=''
                 self.DEBUG(self.DBG,'Disconnect detected','stop')
                 for i in reversed(self.disconnect_handlers):
+                        self.DEBUG(self.DBG, 'Calling disc handler %s' % i, 'stop')
                         i()
                 if self.__dict__.has_key('NonBlockingRoster'):
@@ -95 +120 @@
                         self.NonBlockingTcp.PlugOut()
                 
-        def reconnectAndReauth(self):
-                ''' Just disconnect. We do reconnecting in connection.py '''
-                self.disconnect()
-                return '' 
-
-        def connect(self,server=None,proxy=None, ssl=None, on_stream_start = None):
-                ''' Make a tcp/ip connection, protect it with tls/ssl if possible and start XMPP stream. '''
-                if not server: 
-                        server = (self.Server, self.Port)
-                self._Server,  self._Proxy, self._Ssl = server ,  proxy, ssl
-                self.on_stream_start = on_stream_start
+
+        def send(self, stanza, is_message = False, now = False):
+                ''' interface for putting stanzas on wire. Puts ID to stanza if needed and
+                sends it via socket wrapper'''
+                (id, stanza_to_send) = self.Dispatcher.assign_id(stanza)
+
+                if is_message:
+                        # somehow zeroconf-specific
+                        self.Connection.send(stanza_to_send, True, now = now)
+                else:
+                        self.Connection.send(stanza_to_send, now = now)
+                return id
+
+
+
+        def connect(self, on_connect, on_connect_failure, on_proxy_failure=None, proxy=None, secure=None):
+                ''' 
+                Open XMPP connection (open streams in both directions).
+                :param on_connect: called after stream is successfully opened
+                :param on_connect_failure: called when error occures during connection
+                :param on_proxy_failure: called if error occurres during TCP connection to
+                        proxy server or during connection to the proxy
+                :param proxy: dictionary with proxy data. It should contain at least values
+                        for keys 'host' and 'port' - connection details for proxy server and
+                        optionally keys 'user' and 'pass' as proxy credentials
+                :param secure:
+                '''
+                
+                self.on_connect = on_connect
+                self.on_connect_failure=on_connect_failure
+                self.on_proxy_failure = on_proxy_failure
+                self._secure = secure
+                self.Connection = None
+
                 if proxy:
+                        # with proxies, client connects to proxy instead of directly to
+                        # XMPP server from __init__. 
+                        # tcp_server is hostname used for socket connecting
+                        tcp_server=proxy['host']                        
+                        tcp_port=proxy['port']
+                        self._on_tcp_failure = self.on_proxy_failure
                         if proxy.has_key('type'):
+                                if proxy.has_key('user') and proxy.has_key('pass'):
+                                        proxy_creds=(proxy['user'],proxy['pass'])
+                                else:
+                                        proxy_creds=(None, None)
+                                                                                        
                                 type_ = proxy['type']
                                 if type_ == 'socks5':
-                                        self.socket = transports_nb.NBSOCKS5PROXYsocket(
-                                                self._on_connected, self._on_proxy_failure,
-                                                self._on_connected_failure, proxy, server)
+                                        self.socket = transports_nb.NBSOCKS5ProxySocket(
+                                                on_disconnect=self.on_disconnect,
+                                                proxy_creds=proxy_creds,
+                                                xmpp_server=(self.Server, self.Port))
                                 elif type_ == 'http':
-                                        self.socket = transports_nb.NBHTTPPROXYsocket(self._on_connected,
-                                                self._on_proxy_failure, self._on_connected_failure, proxy,
-                                                server)
+                                        self.socket = transports_nb.NBHTTPProxySocket(
+                                                on_disconnect=self.on_disconnect,
+                                                proxy_creds=proxy_creds,
+                                                xmpp_server=(self.Server, self.Port))
+                                elif type_ == 'bosh':
+                                        tcp_server = transports_nb.urisplit(tcp_server)[1]
+                                        self.socket = transports_nb.NonBlockingHttpBOSH(
+                                                on_disconnect=self.on_disconnect,
+                                                bosh_uri = proxy['host'],
+                                                bosh_port = tcp_port)
                         else:
-                                self.socket = transports_nb.NBHTTPPROXYsocket(self._on_connected,
-                                        self._on_proxy_failure, self._on_connected_failure, proxy,
-                                        server)
+                                        self.socket = transports_nb.NBHTTPProxySocket(
+                                                on_disconnect=self.on_disconnect,
+                                                proxy_creds=(None, None),
+                                                xmpp_server=(self.Server, self.Port))
                 else: 
-                        self.connected = 'tcp'
-                        self.socket = transports_nb.NonBlockingTcp(self._on_connected, 
-                                self._on_connected_failure, server)
+                        self._on_tcp_failure = self._on_connect_failure
+                        tcp_server=self.Server
+                        tcp_port=self.Port
+                        self.socket = transports_nb.NonBlockingTcp(on_disconnect = self.on_disconnect)
+
                 self.socket.PlugIn(self)
-                return True
+
+                self._resolve_hostname(
+                        hostname=tcp_server,
+                        port=tcp_port,
+                        on_success=self._try_next_ip,
+                        on_failure=self._on_tcp_failure)
+                        
+                        
+
+        def _resolve_hostname(self, hostname, port, on_success, on_failure):
+                ''' wrapper of getaddinfo call. FIXME: getaddinfo blocks'''
+                try:
+                        self.ip_addresses = socket.getaddrinfo(hostname,port,
+                                socket.AF_UNSPEC,socket.SOCK_STREAM)
+                except socket.gaierror, (errnum, errstr):
+                        on_failure(err_message='Lookup failure for %s:%s - %s %s' % 
+                                 (self.Server, self.Port, errnum, errstr))
+                else:
+                        on_success()
+                
+                
         
-        def get_attrs(self, on_stream_start):
-                self.on_stream_start = on_stream_start
-                self.onreceive(self._on_receive_document_attrs)
-
-        def _on_proxy_failure(self, reason): 
-                if self.on_proxy_failure:
-                        self.on_proxy_failure(reason)
-
-        def _on_connected_failure(self, retry = None): 
+        def _try_next_ip(self, err_message=None):
+                '''iterates over IP addresses from getaddinfo'''
+                if err_message:
+                        self.DEBUG(self.DBG,err_message,'connect')
+                if self.ip_addresses == []:
+                        self._on_tcp_failure(err_message='Run out of hosts for name %s:%s' % 
+                                (self.Server, self.Port))
+                else:
+                        self.current_ip = self.ip_addresses.pop(0)
+                        self.socket.connect(
+                                conn_5tuple=self.current_ip,
+                                on_connect=lambda: self._xmpp_connect(socket_type='tcp'),
+                                on_connect_failure=self._try_next_ip)
+
+
+        def incoming_stream_version(self):
+                ''' gets version of xml stream'''
+                if self.Dispatcher.Stream._document_attrs.has_key('version'):
+                        return self.Dispatcher.Stream._document_attrs['version']
+                else:
+                        return None
+
+        def _xmpp_connect(self, socket_type):
+                self.connected = socket_type
+                self._xmpp_connect_machine()
+
+
+        def _xmpp_connect_machine(self, mode=None, data=None):
+                '''
+                Finite automaton called after TCP connecting. Takes care of stream opening
+                and features tag handling. Calls _on_stream_start when stream is 
+                started, and _on_connect_failure on failure.
+                '''
+                #FIXME: use RegisterHandlerOnce instead of onreceive
+                log.info('=============xmpp_connect_machine() >> mode: %s, data: %s' % (mode,data))
+
+                def on_next_receive(mode):
+                        if mode is None:
+                                self.onreceive(None)
+                        else:
+                                self.onreceive(lambda data:self._xmpp_connect_machine(mode, data))
+
+                if not mode:
+                        dispatcher_nb.Dispatcher().PlugIn(self)
+                        on_next_receive('RECEIVE_DOCUMENT_ATTRIBUTES')
+
+                elif mode == 'FAILURE':
+                        self._on_connect_failure(err_message='During XMPP connect: %s' % data)
+
+                elif mode == 'RECEIVE_DOCUMENT_ATTRIBUTES':
+                        if data:
+                                self.Dispatcher.ProcessNonBlocking(data)
+                        if not hasattr(self, 'Dispatcher') or \
+                                self.Dispatcher.Stream._document_attrs is None:
+                                self._xmpp_connect_machine(
+                                        mode='FAILURE',
+                                        data='Error on stream open')
+                        if self.incoming_stream_version() == '1.0':
+                                if not self.Dispatcher.Stream.features: 
+                                        on_next_receive('RECEIVE_STREAM_FEATURES')
+                                else:
+                                        self._xmpp_connect_machine(mode='STREAM_STARTED')
+
+                        else:
+                                self._xmpp_connect_machine(mode='STREAM_STARTED')
+
+                elif mode == 'RECEIVE_STREAM_FEATURES':
+                        if data:
+                                # sometimes <features> are received together with document
+                                # attributes and sometimes on next receive...
+                                self.Dispatcher.ProcessNonBlocking(data)
+                        if not self.Dispatcher.Stream.features: 
+                                self._xmpp_connect_machine(
+                                        mode='FAILURE',
+                                        data='Missing <features> in 1.0 stream')
+                        else:
+                                self._xmpp_connect_machine(mode='STREAM_STARTED')
+
+                elif mode == 'STREAM_STARTED':
+                        self._on_stream_start()
+
+        def _on_stream_start(self):
+                '''Called when stream is opened. To be overriden in derived classes.'''
+
+        def _on_connect_failure(self, retry=None, err_message=None): 
+                self.connected = None
+                if err_message:
+                        self.DEBUG(self.DBG, err_message, 'connecting')
                 if self.socket:
                         self.socket.disconnect()
-                if self.on_connect_failure:
-                        self.on_connect_failure(retry)
-
-        def _on_connected(self):
-                # FIXME: why was this needed? Please note that we're working
-                # in nonblocking mode, and this handler is actually called
-                # as soon as connection is initiated, NOT when connection
-                # succeeds, as the name suggests.
-                # # connect succeeded, so no need of this callback anymore 
-                # self.on_connect_failure = None
-                self.connected = 'tcp'
-                if self._Ssl:
-                        transports_nb.NonBlockingTLS().PlugIn(self, now=1)
-                        if not self.Connection: # ssl error, stream is closed
-                                return
-                        self.connected = 'ssl'
-                self.onreceive(self._on_receive_document_attrs)
-                dispatcher_nb.Dispatcher().PlugIn(self)
-                
-        def _on_receive_document_attrs(self, data):
-                if data:
-                        self.Dispatcher.ProcessNonBlocking(data)
-                if not hasattr(self, 'Dispatcher') or \
-                        self.Dispatcher.Stream._document_attrs is None:
-                        return
+                self.on_connect_failure(retry)
+
+        def _on_connect(self):
                 self.onreceive(None)
-                if self.Dispatcher.Stream._document_attrs.has_key('version') and \
-                        self.Dispatcher.Stream._document_attrs['version'] == '1.0':
-                                self.onreceive(self._on_receive_stream_features)
-                                return
-                if self.on_stream_start:
-                        self.on_stream_start()
-                        self.on_stream_start = None
-                return True
-        
-        def _on_receive_stream_features(self, data):