Changeset 9897

Show
Ignore:
Timestamp:
07/08/08 01:04:10 (5 months ago)
Author:
tomk
Message:

new BOSHDispatcher (in dispatcher_nb), improved BOSHClient class, minor changes in other xmpp modules

Location:
branches/bosh_support/src/common/xmpp
Files:
7 modified

Legend:

Unmodified
Added
Removed

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

    r9877 r9897  
    174174                        log.info('Successfully authenticated with remote server.') 
    175175                        handlers=self._owner.Dispatcher.dumpHandlers() 
    176                         print '6' * 79 
    177                         print handlers 
    178                         print '6' * 79 
    179176                        self._owner.Dispatcher.PlugOut() 
    180                         dispatcher_nb.Dispatcher().PlugIn(self._owner) 
     177                        dispatcher_nb.Dispatcher().PlugIn(self._owner, after_SASL=True) 
    181178                        self._owner.Dispatcher.restoreHandlers(handlers) 
    182179                        self._owner.User = self.username 

  • branches/bosh_support/src/common/xmpp/bosh.py

    r9877 r9897  
    11 
    2 import protocol, simplexml, locale, random, dispatcher_nb  
     2import protocol, locale, random, dispatcher_nb  
    33from client_nb import NBCommonClient 
     4import transports_nb 
    45import logging 
     6from simplexml import Node 
    57log = logging.getLogger('gajim.c.x.bosh') 
    68 
    … …  
    810class BOSHClient(NBCommonClient): 
    911        ''' 
    10         Client class implementing BOSH.  
     12        Client class implementing BOSH. Extends common XMPP   
    1113        ''' 
    12         def __init__(self, *args, **kw): 
     14        def __init__(self, domain, idlequeue, caller=None): 
    1315                '''Preceeds constructor of NBCommonClient and sets some of values that will 
    1416                be used as attributes in <body> tag''' 
    15                 self.Namespace = protocol.NS_HTTP_BIND 
    16                 # BOSH parameters should be given via Advanced Configuration Editor 
    17                 self.bosh_xml_lang = None 
    18                 self.bosh_hold = 1 
    19                 self.bosh_wait=60 
    20                 self.bosh_rid=None 
    2117                self.bosh_sid=None 
    22  
    23                 self.bosh_httpversion = 'HTTP/1.1' 
    24                 NBCommonClient.__init__(self, *args, **kw) 
    25  
    26  
    27         def connect(self, *args, **kw): 
    28  
    29  
    30                 if locale.getdefaultlocale()[0]: 
    31                         self.bosh_xml_lang = locale.getdefaultlocale()[0].split('_')[0] 
    3218 
    3319                # with 50-bit random initial rid, session would have to go up 
    … …  
    3723                r.seed() 
    3824                self.bosh_rid = r.getrandbits(50) 
    39  
    40                 proxy = kw['proxy'] 
    41                 #self.bosh_protocol, self.bosh_host, self.bosh_uri = transports_nb.urisplit(proxy['host']) 
    42                 self.bosh_port = proxy['port'] 
     25                self.bosh_sid = None 
     26 
     27                if locale.getdefaultlocale()[0]: 
     28                        self.bosh_xml_lang = locale.getdefaultlocale()[0].split('_')[0] 
     29                else: 
     30                        self.bosh_xml_lang = 'en' 
     31 
     32                self.http_version = 'HTTP/1.1' 
     33                self.bosh_to = domain 
     34 
     35                #self.Namespace = protocol.NS_HTTP_BIND 
     36                #self.defaultNamespace = self.Namespace 
     37                self.bosh_session_on = False 
     38 
     39                NBCommonClient.__init__(self, domain, idlequeue, caller) 
     40 
     41 
     42 
     43        def connect(self, on_connect, on_connect_failure, proxy, hostname=None, port=5222,  
     44                on_proxy_failure=None, secure=None): 
     45                '''  
     46                Open XMPP connection (open XML streams in both directions). 
     47                :param hostname: hostname of XMPP server from SRV request  
     48                :param port: port number of XMPP server 
     49                :param on_connect: called after stream is successfully opened 
     50                :param on_connect_failure: called when error occures during connection 
     51                :param on_proxy_failure: called if error occurres during TCP connection to 
     52                        proxy server or during connection to the proxy 
     53                :param proxy: dictionary with bosh-related paramters. It should contain at  
     54                        least values for keys 'host' and 'port' - connection details for proxy 
     55                        server and optionally keys 'user' and 'pass' as proxy credentials 
     56                :param secure: if  
     57                ''' 
     58                NBCommonClient.connect(self, on_connect, on_connect_failure, hostname, port, 
     59                        on_proxy_failure, proxy, secure) 
     60 
     61                if hostname: 
     62                        self.route_host = hostname 
     63                else: 
     64                        self.route_host = self.Server 
     65 
     66                assert(proxy.has_key('type')) 
     67                assert(proxy['type']=='bosh') 
     68 
    4369                self.bosh_wait = proxy['bosh_wait'] 
    4470                self.bosh_hold = proxy['bosh_hold'] 
    45                 self.bosh_to = proxy['to'] 
    46                 #self.bosh_ack = proxy['bosh_ack'] 
    47                 #self.bosh_secure = proxy['bosh_secure'] 
    48                 NBCommonClient.connect(self, *args, **kw) 
    49                  
    50         def send(self, stanza, now = False): 
    51                 (id, stanza_to_send) = self.Dispatcher.assign_id(stanza) 
    52  
    53                 self.Connection.send( 
    54                         self.boshify_stanza(stanza_to_send), 
    55                         now = now) 
    56                 return id 
    57  
    58         def get_bodytag(self): 
    59                 # this should be called not until after session creation response so sid has 
    60                 # to be initialized.  
    61                 assert(self.sid is not None) 
    62                 self.rid = self.rid+1 
    63                 return protocol.BOSHBody( 
    64                         attrs={ 'rid': str(self.bosh_rid), 
    65                                 'sid': self.bosh_sid}) 
    66  
    67  
    68         def get_initial_bodytag(self): 
    69                 return protocol.BOSHBody( 
    70                         attrs={'content': 'text/xml; charset=utf-8', 
    71                                 'hold': str(self.bosh_hold), 
    72                                 'to': self.bosh_to, 
    73                                 'wait': str(self.bosh_wait), 
    74                                 'rid': str(self.bosh_rid), 
    75                                 'xmpp:version': '1.0', 
    76                                 'xmlns:xmpp': 'urn:xmpp:xbosh'} 
    77                         ) 
    78  
    79         def get_closing_bodytag(self): 
    80                 closing_bodytag = self.get_bodytag() 
    81                 closing_bodytag.setAttr('type', 'terminate') 
    82                 return closing_bodytag 
    83  
    84  
    85         def boshify_stanza(self, stanza): 
    86                 ''' wraps stanza by body tag or modifies message entirely (in case of stream 
    87                 opening and closing''' 
    88                 log.info('boshify_staza - type is: %s' % type(stanza)) 
    89                 if isinstance(stanza, simplexml.Node): 
    90                         tag = self.get_bodytag() 
    91                         return tag.setPayload(stanza) 
    92                 else: 
    93                         # only stream initialization and stream terminatoion are not Nodes 
    94                         if stanza.startswith(dispatcher_nb.XML_DECLARATION): 
    95                                 # stream init 
    96                                 return self.get_initial_bodytag() 
     71                self.bosh_host = proxy['host'] 
     72                self.bosh_port = proxy['port'] 
     73                self.bosh_content = proxy['bosh_content'] 
     74 
     75                # _on_tcp_failure is callback for errors which occur during name resolving or 
     76                # TCP connecting. 
     77                self._on_tcp_failure = self.on_proxy_failure 
     78 
     79 
     80                                                                 
     81                # in BOSH, client connects to Connection Manager instead of directly to 
     82                # XMPP server ((hostname, port)). If HTTP Proxy is specified, client connects 
     83                # to HTTP proxy and Connection Manager is specified at URI and Host header 
     84                # in HTTP message 
     85                                 
     86                # tcp_host, tcp_port is hostname and port for socket connection - Connection 
     87                # Manager or HTTP proxy 
     88                if proxy.has_key('proxy_host') and proxy['proxy_host'] and \ 
     89                        proxy.has_key('proxy_port') and proxy['proxy_port']: 
     90                         
     91                        tcp_host=proxy['proxy_host'] 
     92                        tcp_port=proxy['proxy_port'] 
     93 
     94                        # user and password for HTTP proxy 
     95                        if proxy.has_key('user') and proxy['user'] and \ 
     96                                proxy.has_key('pass') and proxy['pass']: 
     97 
     98                                proxy_creds=(proxy['user'],proxy['pass']) 
    9799                        else: 
    98                                 # should be stream closing 
    99                                 assert(stanza == dispatcher_nb.STREAM_TERMINATOR) 
    100                                 return self.get_closing_bodytag() 
    101  
    102  
     100                                proxy_creds=(None, None) 
     101 
     102                else: 
     103                        tcp_host = transports_nb.urisplit(proxy['host'])[1] 
     104                        tcp_port=proxy['port'] 
     105 
     106                        if tcp_host is None: 
     107                                self._on_connect_failure("Invalid BOSH URI") 
     108                                return 
     109 
     110                self.socket = self.get_socket() 
     111 
     112                self._resolve_hostname( 
     113                        hostname=tcp_host, 
     114                        port=tcp_port, 
     115                        on_success=self._try_next_ip, 
     116                        on_failure=self._on_tcp_failure) 
    103117 
    104118        def _on_stream_start(self): 
    105119                ''' 
    106                 Called after XMPP stream is opened. In BOSH, TLS is negotiated elsewhere  
    107                 so success callback can be invoked. 
     120                Called after XMPP stream is opened. In BOSH, TLS is negotiated on socket 
     121                connect so success callback can be invoked after TCP connect. 
    108122                (authentication is started from auth() method) 
    109123                ''' 
    … …  
    111125                if self.connected == 'tcp': 
    112126                        self._on_connect() 
     127 
     128        def get_socket(self): 
     129                tmp = transports_nb.NonBlockingHTTP( 
     130                        raise_event=self.raise_event, 
     131                        on_disconnect=self.on_http_disconnect, 
     132                        http_uri = self.bosh_host,                       
     133                        http_port = self.bosh_port, 
     134                        http_version = self.http_version 
     135                        ) 
     136                tmp.PlugIn(self) 
     137                return tmp 
     138 
     139        def on_http_disconnect(self): 
     140                log.info('HTTP socket disconnected') 
     141                #import traceback 
     142                #traceback.print_stack() 
     143                if self.bosh_session_on: 
     144                        self.socket.connect( 
     145                                conn_5tuple=self.current_ip, 
     146                                on_connect=self.on_http_reconnect, 
     147                                on_connect_failure=self.on_disconnect) 
     148                else: 
     149                        self.on_disconnect() 
     150 
     151        def on_http_reconnect(self): 
     152                self.socket._plug_idle() 
     153                log.info('Connected to BOSH CM again') 
     154                pass 
     155 
     156 
     157        def on_http_reconnect_fail(self): 
     158                log.error('Error when reconnecting to BOSH CM') 
     159                self.on_disconnect() 
     160                 
     161        def send(self, stanza, now = False): 
     162                (id, stanza_to_send) = self.Dispatcher.assign_id(stanza) 
     163 
     164                self.socket.send( 
     165                        self.boshify_stanza(stanza_to_send), 
     166                        now = now) 
     167                return id 
     168 
     169        def get_rid(self): 
     170                # does this need a lock??" 
     171                self.bosh_rid = self.bosh_rid + 1 
     172                return str(self.bosh_rid) 
     173 
     174        def get_bodytag(self): 
     175                # this should be called not until after session creation response so sid has 
     176                # to be initialized.  
     177                assert(hasattr(self, 'bosh_sid')) 
     178                return protocol.BOSHBody( 
     179                        attrs={ 'rid': self.get_rid(), 
     180                                'sid': self.bosh_sid}) 
     181 
     182        def get_initial_bodytag(self, after_SASL=False): 
     183                tag = protocol.BOSHBody( 
     184                        attrs={'content': self.bosh_content, 
     185                                'hold': str(self.bosh_hold), 
     186                                'route': '%s:%s' % (self.route_host, self.Port), 
     187                                'to': self.bosh_to, 
     188                                'wait': str(self.bosh_wait), 
     189                                'rid': self.get_rid(), 
     190                                'xml:lang': self.bosh_xml_lang, 
     191                                'xmpp:version': '1.0', 
     192                                'ver': '1.6', 
     193                                'xmlns:xmpp': 'urn:xmpp:xbosh'}) 
     194                if after_SASL: 
     195                        tag.delAttr('content') 
     196                        tag.delAttr('hold') 
     197                        tag.delAttr('route') 
     198                        tag.delAttr('wait') 
     199                        tag.delAttr('ver') 
     200                        # xmpp:restart attribute is essential for stream restart request 
     201                        tag.setAttr('xmpp:restart','true') 
     202                        tag.setAttr('sid',self.bosh_sid) 
     203 
     204                return tag 
     205 
     206 
     207        def get_closing_bodytag(self): 
     208                closing_bodytag = self.get_bodytag() 
     209                closing_bodytag.setAttr('type', 'terminate') 
     210                return closing_bodytag 
     211 
     212 
     213        def boshify_stanza(self, stanza=None, body_attrs=None): 
     214                ''' wraps stanza by body tag with rid and sid ''' 
     215                #log.info('boshify_staza - type is: %s, stanza is %s' % (type(stanza), stanza)) 
     216                tag = self.get_bodytag() 
     217                tag.setPayload([stanza]) 
     218                return tag 
     219 
     220 
     221        def on_bodytag_attrs(self, body_attrs): 
     222                #log.info('on_bodytag_attrs: %s' % body_attrs) 
     223                if body_attrs.has_key('type'): 
     224                        if body_attrs['type']=='terminated': 
     225                                # BOSH session terminated  
     226                                self.bosh_session_on = False 
     227                        elif body_attrs['type']=='error': 
     228                                # recoverable error 
     229                                pass 
     230                if not self.bosh_sid: 
     231                        # initial response - when bosh_sid is set 
     232                        self.bosh_session_on = True 
     233                        self.bosh_sid = body_attrs['sid'] 
     234                        self.Dispatcher.Stream._document_attrs['id']=body_attrs['authid'] 
     235 

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

    r9877 r9897  
    4242                         
    4343                ''' 
    44                  
    4544                self.Namespace = protocol.NS_CLIENT 
    46  
     45                self.defaultNamespace = self.Namespace 
     46                 
    4747                self.idlequeue = idlequeue 
    48                 self.defaultNamespace = self.Namespace 
    4948                self.disconnect_handlers = [] 
    5049 
    … …  
    8685                if self.__dict__.has_key('NonBlockingTLS'): 
    8786                        self.NonBlockingTLS.PlugOut() 
    88                 if self.__dict__.has_key('NBHTTPPROXYsocket'): 
     87                if self.__dict__.has_key('NBHTTPProxySocket'): 
    8988                        self.NBHTTPPROXYsocket.PlugOut() 
    90                 if self.__dict__.has_key('NBSOCKS5PROXYsocket'): 
     89                if self.__dict__.has_key('NBSOCKS5ProxySocket'): 
    9190                        self.NBSOCKS5PROXYsocket.PlugOut() 
    92                 if self.__dict__.has_key('NonBlockingTcp'): 
    93                         self.NonBlockingTcp.PlugOut() 
     91                if self.__dict__.has_key('NonBlockingTCP'): 
     92                        self.NonBlockingTCP.PlugOut() 
     93                if self.__dict__.has_key('NonBlockingHTTP'): 
     94                        self.NonBlockingHTTP.PlugOut() 
    9495                 
    9596 
    … …  
    107108                on_proxy_failure=None, proxy=None, secure=None): 
    108109                '''  
    109                 Open XMPP connection (open streams in both directions). 
     110                Open XMPP connection (open XML streams in both directions). 
    110111                :param hostname: hostname of XMPP server from SRV request  
    111112                :param port: port number of XMPP server 
    … …  
    119120                :param secure: 
    120121                ''' 
    121                 self.Port = port 
    122                 if hostname: 
    123                         xmpp_hostname = hostname 
    124                 else: 
    125                         xmpp_hostname = self.Server 
    126  
    127122                self.on_connect = on_connect 
    128123                self.on_connect_failure=on_connect_failure 
    … …  
    130125                self._secure = secure 
    131126                self.Connection = None 
     127                self.Port = port 
     128 
     129 
     130                         
     131                         
     132 
     133        def _resolve_hostname(self, hostname, port, on_success, on_failure): 
     134                ''' wrapper of getaddinfo call. FIXME: getaddinfo blocks''' 
     135                try: 
     136                        self.ip_addresses = socket.getaddrinfo(hostname,port, 
     137                                socket.AF_UNSPEC,socket.SOCK_STREAM) 
     138                except socket.gaierror, (errnum, errstr): 
     139                        on_failure(err_message='Lookup failure for %s:%s - %s %s' %  
     140                                 (self.Server, self.Port, errnum, errstr)) 
     141                else: 
     142                        on_success() 
     143                 
     144                 
     145         
     146        def _try_next_ip(self, err_message=None): 
     147                '''iterates over IP addresses from getaddinfo''' 
     148                if err_message: 
     149                        log.debug('While looping over DNS A records: %s' % connect) 
     150                if self.ip_addresses == []: 
     151                        self._on_tcp_failure(err_message='Run out of hosts for name %s:%s' %  
     152                                (self.Server, self.Port)) 
     153                else: 
     154                        self.current_ip = self.ip_addresses.pop(0) 
     155                        self.socket.connect( 
     156                                conn_5tuple=self.current_ip, 
     157                                on_connect=lambda: self._xmpp_connect(socket_type='tcp'), 
     158                                on_connect_failure=self._try_next_ip) 
     159 
     160 
     161        def incoming_stream_version(self): 
     162                ''' gets version of xml stream''' 
     163                if self.Dispatcher.Stream._document_attrs.has_key('version'): 
     164                        return self.Dispatcher.Stream._document_attrs['version'] 
     165                else: 
     166                        return None 
     167 
     168        def _xmpp_connect(self, socket_type): 
     169                self.connected = socket_type 
     170                self._xmpp_connect_machine() 
     171 
     172 
     173        def _xmpp_connect_machine(self, mode=None, data=None): 
     174                ''' 
     175                Finite automaton called after TCP connecting. Takes care of stream opening 
     176                and features tag handling. Calls _on_stream_start when stream is  
     177                started, and _on_connect_failure on failure. 
     178                ''' 
     179                #FIXME: use RegisterHandlerOnce instead of onreceive 
     180                log.info('========xmpp_connect_machine() >> mode: %s, data: %s' % (mode,str(data)[:20] )) 
     181 
     182                def on_next_receive(mode): 
     183                        log.info('setting %s on next receive' % mode) 
     184                        if mode is None: 
     185                                self.onreceive(None) 
     186                        else: 
     187                                self.onreceive(lambda _data:self._xmpp_connect_machine(mode, _data)) 
     188 
     189                if not mode: 
     190                        dispatcher_nb.Dispatcher().PlugIn(self) 
     191                        on_next_receive('RECEIVE_DOCUMENT_ATTRIBUTES') 
     192 
     193                elif mode == 'FAILURE': 
     194                        self._on_connect_failure(err_message='During XMPP connect: %s' % data) 
     195 
     196                elif mode == 'RECEIVE_DOCUMENT_ATTRIBUTES': 
     197                        if data: 
     198                                self.Dispatcher.ProcessNonBlocking(data) 
     199                        if not hasattr(self, 'Dispatcher') or \ 
     200                                self.Dispatcher.Stream._document_attrs is None: 
     201                                self._xmpp_connect_machine( 
     202                                        mode='FAILURE', 
     203                                        data='Error on stream open') 
     204                        if self.incoming_stream_version() == '1.0': 
     205                                if not self.Dispatcher.Stream.features:  
     206                                        on_next_receive('RECEIVE_STREAM_FEATURES') 
     207                                else: 
     208                                        log.info('got STREAM FEATURES in first read') 
     209                                        self._xmpp_connect_machine(mode='STREAM_STARTED') 
     210 
     211                        else: 
     212                                log.info('incoming stream version less than 1.0') 
     213                                self._xmpp_connect_machine(mode='STREAM_STARTED') 
     214 
     215                elif mode == 'RECEIVE_STREAM_FEATURES': 
     216                        if data: 
     217                                # sometimes <features> are received together with document 
     218                                # attributes and sometimes on next receive... 
     219                                self.Dispatcher.ProcessNonBlocking(data) 
     220                        if not self.Dispatcher.Stream.features:  
     221                                self._xmpp_connect_machine( 
     222                                        mode='FAILURE', 
     223                                        data='Missing <features> in 1.0 stream') 
     224                        else: 
     225                                log.info('got STREAM FEATURES in second read') 
     226                                self._xmpp_connect_machine(mode='STREAM_STARTED') 
     227 
     228                elif mode == 'STREAM_STARTED': 
     229                        self._on_stream_start() 
     230 
     231        def _on_stream_start(self): 
     232                '''Called when stream is opened. To be overriden in derived classes.''' 
     233 
     234        def _on_connect_failure(self, retry=None, err_message=None):  
     235                self.connected = None 
     236                if err_message: 
     237                        log.debug('While connecting: %s' % err_message) 
     238                if self.socket: 
     239                        self.socket.disconnect() 
     240                self.on_connect_failure(retry) 
     241 
     242        def _on_connect(self): 
     243                self.onreceive(None) 
     244                self.on_connect(self, self.connected) 
     245 
     246        def raise_event(self, event_type, data): 
     247                log.info('raising event from transport: %s %s' % (event_type,data)) 
     248                if hasattr(self, 'Dispatcher'): 
     249                        self.Dispatcher.Event('', event_type, data) 
     250                 
     251         
     252        # moved from client.CommonClient: 
     253        def RegisterDisconnectHandler(self,handler): 
     254                """ Register handler that will be called on disconnect.""" 
     255                self.disconnect_handlers.append(handler) 
     256 
     257        def UnregisterDisconnectHandler(self,handler): 
     258                """ Unregister handler that is called on disconnect.""" 
     259                self.disconnect_handlers.remove(handler) 
     260 
     261        def DisconnectHandler(self): 
     262                """ Default disconnect handler. Just raises an IOError. 
     263                        If you choosed to use this class in your production client, 
     264                        override this method or at least unregister it. """ 
     265                raise IOError('Disconnected from server.') 
     266 
     267        def get_connect_type(self): 
     268                """ Returns connection state. F.e.: None / 'tls' / 'tcp+non_sasl' . """ 
     269                return self.connected 
     270 
     271        def get_peerhost(self): 
     272                ''' get the ip address of the account, from which is made connection  
     273                to the server , (e.g. me). 
     274                We will create listening socket on the same ip ''' 
     275                if hasattr(self, 'Connection'): 
     276                        return self.Connection._sock.getsockname() 
     277 
     278 
     279        def auth(self, user, password, resource = '', sasl = 1, on_auth = None): 
     280                ''' Authenticate connnection and bind resource. If resource is not provided 
     281                        random one or library name used. ''' 
     282                self._User, self._Password, self._Resource, self._sasl = user, password, resource, sasl 
     283                self.on_auth = on_auth 
     284                self._on_doc_attrs() 
     285                return 
     286         
     287        def _on_old_auth(self, res): 
     288                if res: 
     289                        self.connected += '+old_auth' 
     290                        self.on_auth(self, 'old_auth') 
     291                else: 
     292                        self.on_auth(self, None) 
     293 
     294        def _on_doc_attrs(self): 
     295                if self._sasl:  
     296                        auth_nb.SASL(self._User, self._Password, self._on_start_sasl).PlugIn(self) 
     297                if not self._sasl or self.SASL.startsasl == 'not-supported': 
     298                        if not self._Resource:  
     299                                self._Resource = 'xmpppy' 
     300                        auth_nb.NonBlockingNonSASL(self._User, self._Password, self._Resource, self._on_old_auth).PlugIn(self) 
     301                        return 
     302                self.onreceive(self._on_start_sasl) 
     303                self.SASL.auth() 
     304                return True 
     305                 
     306        def _on_start_sasl(self, data=None): 
     307                if data: 
     308                        self.Dispatcher.ProcessNonBlocking(data) 
     309                if not self.__dict__.has_key('SASL'):  
     310                        # SASL is pluged out, possible disconnect  
     311                        return 
     312                if self.SASL.startsasl == 'in-process':  
     313                        return 
     314                self.onreceive(None) 
     315                if self.SASL.startsasl == 'failure':  
     316                        # wrong user/pass, stop auth 
     317                        self.connected = None 
     318                        self._on_sasl_auth(None) 
     319                        self.SASL.PlugOut() 
     320                elif self.SASL.startsasl == 'success': 
     321                        auth_nb.NonBlockingBind().PlugIn(self) 
     322                        self.onreceive(self._on_auth_bind) 
     323                return True 
     324                 
     325        def _on_auth_bind(self, data): 
     326                if data: 
     327                        self.Dispatcher.ProcessNonBlocking(data) 
     328