signature:
def connect ( self , server = None , port = None ) :
    """Establish a client connection to a server .
 [ component only ]
 : Parameters :
 - ` server ` : name or address of the server to use . If not given
 then use the one specified when creating the object .
 - ` port ` : port number of the server to use . If not given then use
 the one specified when creating the object .
 : Types :
 - ` server ` : ` unicode `
 - ` port ` : ` int `"""

implementation:
self . lock . acquire ( )
try :
    self . _connect ( server , port )
finally :
    self . lock . release ( )


signature:
def _connect ( self , server = None , port = None ) :
    """Same as ` ComponentStream . connect ` but assume ` self . lock ` is acquired ."""

implementation:
if self . me . node or self . me . resource :
    raise Value ( "Component JID may have only domain defined" )
if not server :
    server = self . server
if not port :
    port = self . port
if not server or not port :
    raise ValueError ( "Server or port not given" )
Stream . _connect ( self , server , port , None , self . me )


signature:
def _compute_handshake ( self ) :
    """Compute the authentication handshake value .
 : return : the computed hash value .
 : returntype : ` str `"""

implementation: return hashlib . sha1 ( to_utf8 ( self . stream_id ) + to_utf8 ( self . secret ) ) . hexdigest ( )


signature:
def _auth ( self ) :
    """Authenticate on the server .
 [ component only ]"""

implementation:
if self . authenticated :
    self . __logger . debug ( "_auth: already authenticated" )
    return
self . __logger . debug ( "doing handshake..." )
hash_value = self . _compute_handshake ( )
n = common_root . newTextChild ( None , "handshake" , hash_value )
self . _write_node ( n )
n . unlinkNode ( )
n . freeNode ( )
self . __logger . debug ( "handshake hash sent." )


signature:
def _process_node ( self , node ) :
    """Process first level element of the stream .
 Handle component handshake ( authentication ) element , and
 treat elements in " jabber : component : accept " , " jabber : client "
 and " jabber : server " equally ( pass to ` self . process _ stanza ` ) .
 All other elements are passed to ` Stream . _ process _ node ` .
 : Parameters :
 - ` node ` : XML node describing the element"""

implementation:
ns = node . ns ( )
if ns :
    ns_uri = node . ns ( ) . getContent ( )
if ( not ns or ns_uri == "jabber:component:accept" ) and node . name == "handshake" :
    if self . initiator and not self . authenticated :
        self . authenticated = 1
        self . state_change ( "authenticated" , self . me )
        self . _post_auth ( )
        return
    elif not self . authenticated and node . getContent ( ) == self . _compute_handshake ( ) :
        self . peer = self . me
        n = common_doc . newChild ( None , "handshake" , None )
        self . _write_node ( n )
        n . unlinkNode ( )
        n . freeNode ( )
        self . peer_authenticated = 1
        self . state_change ( "authenticated" , self . peer )
        self . _post_auth ( )
        return
    else :
        self . _send_stream_error ( "not-authorized" )
        raise FatalComponentStreamError ( "Hanshake error." )
if ns_uri in ( "jabber:component:accept" , "jabber:client" , "jabber:server" ) :
    stanza = stanza_factory ( node )
    self . lock . release ( )
    try :
        self . process_stanza ( stanza )
    finally :
        self . lock . acquire ( )
        stanza . free ( )
    return
return Stream . _process_node ( self , node )


signature:
def _set_state ( self , state ) :
    """Set ` _ state ` and notify any threads waiting for the change ."""

implementation:
logger . debug ( " _set_state({0!r})" . format ( state ) )
self . _state = state
self . _state_cond . notify ( )


signature:
def connect ( self , addr , port = None , service = None ) :
    """Start establishing TCP connection with given address .
 One of : ` port ` or ` service ` must be provided and ` addr ` must be
 a domain name and not an IP address if ` port ` is not given .
 When ` service ` is given try an SRV lookup for that service
 at domain ` addr ` . If ` service ` is not given or ` addr ` is an IP address ,
 or the SRV lookup fails , connect to ` port ` at host ` addr ` directly .
 [ initiating entity only ]
 : Parameters :
 - ` addr ` : peer name or IP address
 - ` port ` : port number to connect to
 - ` service ` : service name ( to be resolved using SRV DNS records )"""

implementation:
with self . lock :
    self . _connect ( addr , port , service )


signature:
def _connect ( self , addr , port , service ) :
    """Same as ` connect ` , but assumes ` lock ` acquired ."""

implementation:
self . _dst_name = addr
self . _dst_port = port
family = None
try :
    res = socket . getaddrinfo ( addr , port , socket . AF_UNSPEC , socket . SOCK_STREAM , 0 , socket . AI_NUMERICHOST )
    family = res [ 0 ] [ 0 ]
    sockaddr = res [ 0 ] [ 4 ]
except socket . gaierror :
    family = None
    sockaddr = None
if family is not None :
    if not port :
        raise ValueError ( "No port number given with literal IP address" )
    self . _dst_service = None
    self . _family = family
    self . _dst_addrs = [ ( family , sockaddr ) ]
    self . _set_state ( "connect" )
elif service is not None :
    self . _dst_service = service
    self . _set_state ( "resolve-srv" )
    self . _dst_name = addr
elif port :
    self . _dst_nameports = [ ( self . _dst_name , self . _dst_port ) ]
    self . _dst_service = None
    self . _set_state ( "resolve-hostname" )
else :
    raise ValueError ( "No port number and no SRV service name given" )


signature:
def _resolve_srv ( self ) :
    """Start resolving the SRV record ."""

implementation:
resolver = self . settings [ "dns_resolver" ]
# pylint : disable = W0621
self . _set_state ( "resolving-srv" )
self . event ( ResolvingSRVEvent ( self . _dst_name , self . _dst_service ) )
resolver . resolve_srv ( self . _dst_name , self . _dst_service , "tcp" , callback = self . _got_srv )


signature:
def _got_srv ( self , addrs ) :
    """Handle SRV lookup result .
 : Parameters :
 - ` addrs ` : properly sorted list of ( hostname , port ) tuples"""

implementation:
with self . lock :
    if not addrs :
        self . _dst_service = None
        if self . _dst_port :
            self . _dst_nameports = [ ( self . _dst_name , self . _dst_port ) ]
        else :
            self . _dst_nameports = [ ]
            self . _set_state ( "aborted" )
            raise DNSError ( "Could not resolve SRV for service {0!r}" " on host {1!r} and fallback port number not given" . format ( self . _dst_service , self . _dst_name ) )
    elif addrs == [ ( "." , 0 ) ] :
        self . _dst_nameports = [ ]
        self . _set_state ( "aborted" )
        raise DNSError ( "Service {0!r} not available on host {1!r}" . format ( self . _dst_service , self . _dst_name ) )
    else :
        self . _dst_nameports = addrs
    self . _set_state ( "resolve-hostname" )


signature:
def _resolve_hostname ( self ) :
    """Start hostname resolution for the next name to try .
 [ called with ` lock ` acquired ]"""

implementation:
self . _set_state ( "resolving-hostname" )
resolver = self . settings [ "dns_resolver" ]
# pylint : disable = W0621
logger . debug ( "_dst_nameports: {0!r}" . format ( self . _dst_nameports ) )
name , port = self . _dst_nameports . pop ( 0 )
self . _dst_hostname = name
resolver . resolve_address ( name , callback = partial ( self . _got_addresses , name , port ) , allow_cname = self . _dst_service is None )
self . event ( ResolvingAddressEvent ( name ) )


signature:
def _got_addresses ( self , name , port , addrs ) :
    """Handler DNS address record lookup result .
 : Parameters :
 - ` name ` : the name requested
 - ` port ` : port number to connect to
 - ` addrs ` : list of ( family , address ) tuples"""

implementation:
with self . lock :
    if not addrs :
        if self . _dst_nameports :
            self . _set_state ( "resolve-hostname" )
            return
        else :
            self . _dst_addrs = [ ]
            self . _set_state ( "aborted" )
            raise DNSError ( "Could not resolve address record for {0!r}" . format ( name ) )
    self . _dst_addrs = [ ( family , ( addr , port ) ) for ( family , addr ) in addrs ]
    self . _set_state ( "connect" )


signature:
def _start_connect ( self ) :
    """Start connecting to the next address on the ` _ dst _ addrs ` list .
 [ called with ` lock ` acquired ]"""

implementation:
family , addr = self . _dst_addrs . pop ( 0 )
self . _socket = socket . socket ( family , socket . SOCK_STREAM )
self . _socket . setblocking ( False )
self . _dst_addr = addr
self . _family = family
try :
    self . _socket . connect ( addr )
except socket . error , err :
    logger . debug ( "Connect error: {0}" . format ( err ) )
    if err . args [ 0 ] in BLOCKING_ERRORS :
        self . _set_state ( "connecting" )
        self . _write_queue . append ( ContinueConnect ( ) )
        self . _write_queue_cond . notify ( )
        self . event ( ConnectingEvent ( addr ) )
        return
    elif self . _dst_addrs :
        self . _set_state ( "connect" )
        return
    elif self . _dst_nameports :
        self . _set_state ( "resolve-hostname" )
        return
    else :
        self . _socket . close ( )
        self . _socket = None
        self . _set_state ( "aborted" )
        self . _write_queue . clear ( )
        self . _write_queue_cond . notify ( )
        raise
self . _connected ( )


signature:
def _connected ( self ) :
    """Handle connection success ."""

implementation:
self . _auth_properties [ 'remote-ip' ] = self . _dst_addr [ 0 ]
if self . _dst_service :
    self . _auth_properties [ 'service-domain' ] = self . _dst_name
if self . _dst_hostname is not None :
    self . _auth_properties [ 'service-hostname' ] = self . _dst_hostname
else :
    self . _auth_properties [ 'service-hostname' ] = self . _dst_addr [ 0 ]
self . _auth_properties [ 'security-layer' ] = None
self . event ( ConnectedEvent ( self . _dst_addr ) )
self . _set_state ( "connected" )
self . _stream . transport_connected ( )


signature:
def _continue_connect ( self ) :
    """Continue connecting .
 [ called with ` lock ` acquired ]
 : Return : ` True ` when just connected"""

implementation:
try :
    self . _socket . connect ( self . _dst_addr )
except socket . error , err :
    logger . debug ( "Connect error: {0}" . format ( err ) )
    if err . args [ 0 ] == errno . EISCONN :
        pass
    elif err . args [ 0 ] in BLOCKING_ERRORS :
        return None
    elif self . _dst_addrs :
        self . _set_state ( "connect" )
        return None
    elif self . _dst_nameports :
        self . _set_state ( "resolve-hostname" )
        return None
    else :
        self . _socket . close ( )
        self . _socket = None
        self . _set_state ( "aborted" )
        raise
self . _connected ( )


signature:
def _write ( self , data ) :
    """Write raw data to the socket .
 : Parameters :
 - ` data ` : data to send
 : Types :
 - ` data ` : ` bytes `"""

implementation:
OUT_LOGGER . debug ( "OUT: %r" , data )
if self . _hup or not self . _socket :
    raise PyXMPPIOError ( u"Connection closed." )
try :
    while data :
        try :
            sent = self . _socket . send ( data )
        except ssl . SSLError , err :
            if err . args [ 0 ] == ssl . SSL_ERROR_WANT_WRITE :
                continue
            else :
                raise
        except socket . error , err :
            if err . args [ 0 ] == errno . EINTR :
                continue
            if err . args [ 0 ] in BLOCKING_ERRORS :
                wait_for_write ( self . _socket )
                continue
            raise
        data = data [ sent : ]
except ( IOError , OSError , socket . error ) , err :
    raise PyXMPPIOError ( u"IO Error: {0}" . format ( err ) )


signature:
def set_target ( self , stream ) :
    """Make the ` stream ` the target for this transport instance .
 The ' stream _ start ' , ' stream _ end ' and ' stream _ element ' methods
 of the target will be called when appropriate content is received .
 : Parameters :
 - ` stream ` : the stream handler to receive stream content
 from the transport
 : Types :
 - ` stream ` : ` StreamBase `"""

implementation:
with self . lock :
    if self . _stream :
        raise ValueError ( "Target stream already set" )
    self . _stream = stream
    self . _reader = StreamReader ( stream )


signature:
def send_stream_head ( self , stanza_namespace , stream_from , stream_to , stream_id = None , version = u'1.0' , language = None ) :
    """Send stream head via the transport .
 : Parameters :
 - ` stanza _ namespace ` : namespace of stream stanzas ( e . g .
 ' jabber : client ' )
 - ` stream _ from ` : the ' from ' attribute of the stream . May be ` None ` .
 - ` stream _ to ` : the ' to ' attribute of the stream . May be ` None ` .
 - ` version ` : the ' version ' of the stream .
 - ` language ` : the ' xml : lang ' of the stream
 : Types :
 - ` stanza _ namespace ` : ` unicode `
 - ` stream _ from ` : ` unicode `
 - ` stream _ to ` : ` unicode `
 - ` version ` : ` unicode `
 - ` language ` : ` unicode `"""

implementation:
# pylint : disable = R0913
with self . lock :
    self . _serializer = XMPPSerializer ( stanza_namespace , self . settings [ "extra_ns_prefixes" ] )
    head = self . _serializer . emit_head ( stream_from , stream_to , stream_id , version , language )
    self . _write ( head . encode ( "utf-8" ) )


signature:
def send_stream_tail ( self ) :
    """Send stream tail via the transport ."""

implementation:
with self . lock :
    if not self . _socket or self . _hup :
        logger . debug ( u"Cannot send stream closing tag: already closed" )
        return
    data = self . _serializer . emit_tail ( )
    try :
        self . _write ( data . encode ( "utf-8" ) )
    except ( IOError , SystemError , socket . error ) , err :
        logger . debug ( u"Sending stream closing tag failed: {0}" . format ( err ) )
    self . _serializer = None
    self . _hup = True
    if self . _tls_state is None :
        try :
            self . _socket . shutdown ( socket . SHUT_WR )
        except socket . error :
            pass
    self . _set_state ( "closing" )
    self . _write_queue . clear ( )
    self . _write_queue_cond . notify ( )


signature:
def send_element ( self , element ) :
    """Send an element via the transport ."""

implementation:
with self . lock :
    if self . _eof or self . _socket is None or not self . _serializer :
        logger . debug ( "Dropping element: {0}" . format ( element_to_unicode ( element ) ) )
        return
    data = self . _serializer . emit_stanza ( element )
    self . _write ( data . encode ( "utf-8" ) )


signature:
def prepare ( self ) :
    """When connecting start the next connection step and schedule
 next ` prepare ` call , when connected return ` HandlerReady ( ) `"""

implementation:
result = HandlerReady ( )
logger . debug ( "TCPTransport.prepare(): state: {0!r}" . format ( self . _state ) )
with self . lock :
    if self . _state in ( "connected" , "closing" , "closed" , "aborted" ) : # no need to call prepare ( ) . fileno ( ) is stable
        pass
    elif self . _state == "connect" :
        self . _start_connect ( )
        result = PrepareAgain ( None )
    elif self . _state == "resolve-hostname" :
        self . _resolve_hostname ( )
        result = PrepareAgain ( 0 )
    elif self . _state == "resolve-srv" :
        self . _resolve_srv ( )
        result = PrepareAgain ( 0 )
    else : # wait for i / o , but keep calling prepare ( )
        result = PrepareAgain ( None )
logger . debug ( "TCPTransport.prepare(): new state: {0!r}" . format ( self . _state ) )
return result


signature:
def is_readable ( self ) :
    """: Return : ` True ` when the I / O channel can be read"""

implementation: return self . _socket is not None and not self . _eof and ( self . _state in ( "connected" , "closing" ) or self . _state == "tls-handshake" and self . _tls_state == "want_read" )


signature:
def wait_for_readability ( self ) :
    """Stop current thread until the channel is readable .
 : Return : ` False ` if it won ' t be readable ( e . g . is closed )"""

implementation:
with self . lock :
    while True :
        if self . _socket is None or self . _eof :
            return False
        if self . _state in ( "connected" , "closing" ) :
            return True
        if self . _state == "tls-handshake" and self . _tls_state == "want_read" :
            return True
        self . _state_cond . wait ( )


signature:
def wait_for_writability ( self ) :
    """Stop current thread until the channel is writable .
 : Return : ` False ` if it won ' t be readable ( e . g . is closed )"""

implementation:
with self . lock :
    while True :
        if self . _state in ( "closing" , "closed" , "aborted" ) :
            return False
        if self . _socket and bool ( self . _write_queue ) :
            return True
        self . _write_queue_cond . wait ( )
return False


signature:
def handle_write ( self ) :
    """Handle the ' channel writable ' state . E . g . send buffered data via a
 socket ."""

implementation:
with self . lock :
    logger . debug ( "handle_write: queue: {0!r}" . format ( self . _write_queue ) )
    try :
        job = self . _write_queue . popleft ( )
    except IndexError :
        return
    if isinstance ( job , WriteData ) :
        self . _do_write ( job . data )
        # pylint : disable = E1101
    elif isinstance ( job , ContinueConnect ) :
        self . _continue_connect ( )
    elif isinstance ( job , StartTLS ) :
        self . _initiate_starttls ( ** job . kwargs )
    elif isinstance ( job , TLSHandshake ) :
        self . _continue_tls_handshake ( )
    else :
        raise ValueError ( "Unrecognized job in the write queue: " "{0!r}" . format ( job ) )


signature:
def starttls ( self , ** kwargs ) :
    """Request a TLS handshake on the socket ans switch
 to encrypted output .
 The handshake will start after any currently buffered data is sent .
 : Parameters :
 - ` kwargs ` : arguments for : std : ` ssl . wrap _ socket `"""

implementation:
with self . lock :
    self . event ( TLSConnectingEvent ( ) )
    self . _write_queue . append ( StartTLS ( ** kwargs ) )
    self . _write_queue_cond . notify ( )


signature:
def getpeercert ( self ) :
    """Return the peer certificate .
 : ReturnType : ` pyxmpp2 . cert . Certificate `"""

implementation:
with self . lock :
    if not self . _socket or self . _tls_state != "connected" :
        raise ValueError ( "Not TLS-connected" )
    return get_certificate_from_ssl_socket ( self . _socket )


signature:
def _initiate_starttls ( self , ** kwargs ) :
    """Initiate starttls handshake over the socket ."""

implementation:
if self . _tls_state == "connected" :
    raise RuntimeError ( "Already TLS-connected" )
kwargs [ "do_handshake_on_connect" ] = False
logger . debug ( "Wrapping the socket into ssl" )
self . _socket = ssl . wrap_socket ( self . _socket , ** kwargs )
self . _set_state ( "tls-handshake" )
self . _continue_tls_handshake ( )


signature:
def _continue_tls_handshake ( self ) :
    """Continue a TLS handshake ."""

implementation:
try :
    logger . debug ( " do_handshake()" )
    self . _socket . do_handshake ( )
except ssl . SSLError , err :
    if err . args [ 0 ] == ssl . SSL_ERROR_WANT_READ :
        self . _tls_state = "want_read"
        logger . debug ( "   want_read" )
        self . _state_cond . notify ( )
        return
    elif err . args [ 0 ] == ssl . SSL_ERROR_WANT_WRITE :
        self . _tls_state = "want_write"
        logger . debug ( "   want_write" )
        self . _write_queue . appendleft ( TLSHandshake )
        return
    else :
        raise
self . _tls_state = "connected"
self . _set_state ( "connected" )
self . _auth_properties [ 'security-layer' ] = "TLS"
if "tls-unique" in CHANNEL_BINDING_TYPES :
    try : # pylint : disable = E1103
        tls_unique = self . _socket . get_channel_binding ( "tls-unique" )
    except ValueError :
        pass
    else :
        self . _auth_properties [ 'channel-binding' ] = { "tls-unique" : tls_unique }
try :
    cipher = self . _socket . cipher ( )
except AttributeError : # SSLSocket . cipher doesn ' t work on PyPy
    cipher = "unknown"
cert = get_certificate_from_ssl_socket ( self . _socket )
self . event ( TLSConnectedEvent ( cipher , cert ) )


signature:
def handle_read ( self ) :
    """Handle the ' channel readable ' state . E . g . read from a socket ."""

implementation:
with self . lock :
    logger . debug ( "handle_read()" )
    if self . _eof or self . _socket is None :
        return
    if self . _state == "tls-handshake" :
        while True :
            logger . debug ( "tls handshake read..." )
            self . _continue_tls_handshake ( )
            logger . debug ( "  state: {0}" . format ( self . _tls_state ) )
            if self . _tls_state != "want_read" :
                break
    elif self . _tls_state == "connected" :
        while self . _socket and not self . _eof :
            logger . debug ( "tls socket read..." )
            try :
                data = self . _socket . read ( 4096 )
            except ssl . SSLError , err :
                if err . args [ 0 ] == ssl . SSL_ERROR_WANT_READ :
                    break
                elif err . args [ 0 ] == ssl . SSL_ERROR_WANT_WRITE :
                    break
                else :
                    raise
            except socket . error , err :
                if err . args [ 0 ] == errno . EINTR :
                    continue
                elif err . args [ 0 ] in BLOCKING_ERRORS :
                    break
                elif err . args [ 0 ] == errno . ECONNRESET :
                    logger . warning ( "Connection reset by peer" )
                    data = None
                else :
                    raise
            self . _feed_reader ( data )
    else :
        while self . _socket and not self . _eof :
            logger . debug ( "raw socket read..." )
            try :
                data = self . _socket . recv ( 4096 )
            except socket . error , err :
                if err . args [ 0 ] == errno . EINTR :
                    continue
                elif err . args [ 0 ] in BLOCKING_ERRORS :
                    break
                elif err . args [ 0 ] == errno . ECONNRESET :
                    logger . warning ( "Connection reset by peer" )
                    data = None
                else :
                    raise
            self . _feed_reader ( data )


signature:
def handle_hup ( self ) :
    """Handle the ' channel hungup ' state . The handler should not be writable
 after this ."""

implementation:
with self . lock :
    if self . _state == 'connecting' and self . _dst_addrs :
        self . _hup = False
        self . _set_state ( "connect" )
        return
self . _hup = True


signature:
def handle_err ( self ) :
    """Handle an error reported ."""

implementation:
with self . lock :
    if self . _state == 'connecting' and self . _dst_addrs :
        self . _hup = False
        self . _set_state ( "connect" )
        return
    self . _socket . close ( )
    self . _socket = None
    self . _set_state ( "aborted" )
    self . _write_queue . clear ( )
    self . _write_queue_cond . notify ( )
raise PyXMPPIOError ( "Unhandled error on socket" )


signature:
def disconnect ( self ) :
    """Disconnect the stream gracefully ."""

implementation:
logger . debug ( "TCPTransport.disconnect()" )
with self . lock :
    if self . _socket is None :
        if self . _state != "closed" :
            self . event ( DisconnectedEvent ( self . _dst_addr ) )
            self . _set_state ( "closed" )
        return
    if self . _hup or not self . _serializer :
        self . _close ( )
    else :
        self . send_stream_tail ( )


signature:
def _close ( self ) :
    """Same as ` _ close ` but expects ` lock ` acquired ."""

implementation:
if self . _state != "closed" :
    self . event ( DisconnectedEvent ( self . _dst_addr ) )
    self . _set_state ( "closed" )
if self . _socket is None :
    return
try :
    self . _socket . shutdown ( socket . SHUT_RDWR )
except socket . error :
    pass
self . _socket . close ( )
self . _socket = None
self . _write_queue . clear ( )
self . _write_queue_cond . notify ( )


signature:
def _feed_reader ( self , data ) :
    """Feed the stream reader with data received .
 [ called with ` lock ` acquired ]
 If ` data ` is None or empty , then stream end ( peer disconnected ) is
 assumed and the stream is closed .
 ` lock ` is acquired during the operation .
 : Parameters :
 - ` data ` : data received from the stream socket .
 : Types :
 - ` data ` : ` unicode `"""

implementation:
IN_LOGGER . debug ( "IN: %r" , data )
if data :
    self . lock . release ( )
    # not to deadlock with the stream
    try :
        self . _reader . feed ( data )
    finally :
        self . lock . acquire ( )
else :
    self . _eof = True
    self . lock . release ( )
    # not to deadlock with the stream
    try :
        self . _stream . stream_eof ( )
    finally :
        self . lock . acquire ( )
    if not self . _serializer :
        if self . _state != "closed" :
            self . event ( DisconnectedEvent ( self . _dst_addr ) )
            self . _set_state ( "closed" )


signature:
def event ( self , event ) :
    """Pass an event to the target stream or just log it ."""

implementation:
logger . debug ( u"TCP transport event: {0}" . format ( event ) )
if self . _stream :
    event . stream = self . _stream
self . _event_queue . put ( event )


signature:
def start ( self , tag , attrs ) :
    """Handle the start tag .
 Call the handler ' s ' stream _ start ' methods with
 an empty root element if it is top level .
 For lower level tags use : etree : ` ElementTree . TreeBuilder ` to collect
 them ."""

implementation:
if self . _level == 0 :
    self . _root = ElementTree . Element ( tag , attrs )
    self . _handler . stream_start ( self . _root )
if self . _level < 2 :
    self . _builder = ElementTree . TreeBuilder ( )
self . _level += 1
return self . _builder . start ( tag , attrs )


signature:
def end ( self , tag ) :
    """Handle an end tag .
 Call the handler ' s ' stream _ end ' method with
 an the root element ( built by the ` start ` method ) .
 On the first level below root , sent the built element tree
 to the handler via the ' stanza methods ' .
 Any tag below will be just added to the tree builder ."""

implementation:
self . _level -= 1
if self . _level < 0 :
    self . _handler . stream_parse_error ( u"Unexpected end tag for: {0!r}" . format ( tag ) )
    return
if self . _level == 0 :
    if tag != self . _root . tag :
        self . _handler . stream_parse_error ( u"Unexpected end tag for:" " {0!r} (stream end tag expected)" . format ( tag ) )
        return
    self . _handler . stream_end ( )
    return
element = self . _builder . end ( tag )
if self . _level == 1 :
    self . _handler . stream_element ( element )


signature:
def feed ( self , data ) :
    """Feed the parser with a chunk of data . Apropriate methods
 of ` handler ` will be called whenever something interesting is
 found .
 : Parameters :
 - ` data ` : the chunk of data to parse .
 : Types :
 - ` data ` : ` str `"""

implementation:
with self . lock :
    if self . in_use :
        raise StreamParseError ( "StreamReader.feed() is not reentrant!" )
    self . in_use = True
    try :
        if not self . _started : # workaround for lxml bug when fed with a big chunk at once
            if len ( data ) > 1 :
                self . parser . feed ( data [ : 1 ] )
                data = data [ 1 : ]
            self . _started = True
        if data :
            self . parser . feed ( data )
        else :
            self . parser . close ( )
    except ElementTree . ParseError , err :
        self . handler . stream_parse_error ( unicode ( err ) )
    finally :
        self . in_use = False


signature:
def _from_xml ( self , element ) :
    """Initialize an ErrorElement object from an XML element .
 : Parameters :
 - ` element ` : XML element to be decoded .
 : Types :
 - ` element ` : : etree : ` ElementTree . Element `"""

implementation:
# pylint : disable - msg = R0912
if element . tag != self . error_qname :
    raise ValueError ( u"{0!r} is not a {1!r} element" . format ( element , self . error_qname ) )
lang = element . get ( XML_LANG_QNAME , None )
if lang :
    self . language = lang
self . condition = None
for child in element :
    if child . tag . startswith ( self . cond_qname_prefix ) :
        if self . condition is not None :
            logger . warning ( "Multiple conditions in XMPP error element." )
            continue
        self . condition = deepcopy ( child )
    elif child . tag == self . text_qname :
        lang = child . get ( XML_LANG_QNAME , None )
        if lang :
            self . language = lang
        self . text = child . text . strip ( )
    else :
        bad = False
        for prefix in ( STREAM_QNP , STANZA_CLIENT_QNP , STANZA_SERVER_QNP , STANZA_ERROR_QNP , STREAM_ERROR_QNP ) :
            if child . tag . startswith ( prefix ) :
                logger . warning ( "Unexpected stream-namespaced" " element in error." )
                bad = True
                break
        if not bad :
            self . custom_condition . append ( deepcopy ( child ) )
if self . condition is None :
    self . condition = ElementTree . Element ( self . cond_qname_prefix + "undefined-condition" )
if self . condition . tag in OBSOLETE_CONDITIONS :
    new_cond_name = OBSOLETE_CONDITIONS [ self . condition . tag ]
    self . condition = ElementTree . Element ( new_cond_name )


signature:
def as_xml ( self ) :
    """Return the XML error representation .
 : returntype : : etree : ` ElementTree . Element `"""

implementation:
result = ElementTree . Element ( self . error_qname )
result . append ( deepcopy ( self . condition ) )
if self . text :
    text = ElementTree . SubElement ( result , self . text_qname )
    if self . language :
        text . set ( XML_LANG_QNAME , self . language )
    text . text = self . text
return result


implementation:
ErrorElement . _from_xml ( self , element )
error_type = element . get ( u"type" )
if error_type :
    self . error_type = error_type


signature:
def as_xml ( self , stanza_namespace = None ) : # pylint : disable - msg = W0221
    """Return the XML error representation .
 : Parameters :
 - ` stanza _ namespace ` : namespace URI of the containing stanza
 : Types :
 - ` stanza _ namespace ` : ` unicode `
 : returntype : : etree : ` ElementTree . Element `"""

implementation:
if stanza_namespace :
    self . error_qname = "{{{0}}}error" . format ( stanza_namespace )
    self . text_qname = "{{{0}}}text" . format ( stanza_namespace )
result = ErrorElement . as_xml ( self )
result . set ( "type" , self . error_type )
return result


signature:
def run ( self ) :
    """The thread function .
 First , call the handler ' s ' prepare ' method until it returns
 ` HandlerReady ` then loop waiting for the socket input and calling
 ' handle _ read ' on the handler ."""

implementation:
# pylint : disable - msg = R0912
interval = self . settings [ "poll_interval" ]
prepared = False
timeout = 0.1
while not self . _quit :
    if not prepared :
        logger . debug ( "{0}: preparing handler: {1!r}" . format ( self . name , self . io_handler ) )
        ret = self . io_handler . prepare ( )
        logger . debug ( "{0}: prepare result: {1!r}" . format ( self . name , ret ) )
        if isinstance ( ret , HandlerReady ) :
            prepared = True
        elif isinstance ( ret , PrepareAgain ) :
            if ret . timeout is not None :
                timeout = ret . timeout
        else :
            raise TypeError ( "Unexpected result type from prepare()" )
    if self . io_handler . is_readable ( ) :
        logger . debug ( "{0}: readable" . format ( self . name ) )
        fileno = self . io_handler . fileno ( )
        if fileno is not None :
            readable = wait_for_read ( fileno , interval )
            if readable :
                self . io_handler . handle_read ( )
    elif not prepared :
        if timeout :
            time . sleep ( timeout )
    else :
        logger . debug ( "{0}: waiting for readability" . format ( self . name ) )
        if not self . io_handler . wait_for_readability ( ) :
            break


signature:
def run ( self ) :
    """The thread function .
 Loop waiting for the handler and socket being writable and calling
 ` interfaces . IOHandler . handle _ write ` ."""

implementation:
while not self . _quit :
    interval = self . settings [ "poll_interval" ]
    if self . io_handler . is_writable ( ) :
        logger . debug ( "{0}: writable" . format ( self . name ) )
        fileno = self . io_handler
        if fileno :
            writable = wait_for_write ( fileno , interval )
            if writable :
                self . io_handler . handle_write ( )
    else :
        logger . debug ( "{0}: waiting for writaility" . format ( self . name ) )
        if not self . io_handler . wait_for_writability ( ) :
            break


signature:
def run ( self ) :
    """The thread function . Calls ` self . run ( ) ` and if it raises
 an exception , stores it in self . exc _ info and exc _ queue"""

implementation:
logger . debug ( "{0}: entering thread" . format ( self . name ) )
while True :
    try :
        self . event_dispatcher . loop ( )
    except Exception : # pylint : disable - msg = W0703
        self . exc_info = sys . exc_info ( )
        logger . debug ( u"exception in the {0!r} thread:" . format ( self . name ) , exc_info = self . exc_info )
        if self . exc_queue :
            self . exc_queue . put ( ( self , self . exc_info ) )
            continue
        else :
            logger . debug ( "{0}: aborting thread" . format ( self . name ) )
            return
    except :
        logger . debug ( "{0}: aborting thread" . format ( self . name ) )
        return
    break
logger . debug ( "{0}: exiting thread" . format ( self . name ) )


signature:
def run ( self ) :
    """The thread function ."""

implementation:
# pylint : disable - msg = W0212
timeout = self . method . _pyxmpp_timeout
recurring = self . method . _pyxmpp_recurring
while not self . _quit and timeout is not None :
    if timeout :
        time . sleep ( timeout )
    if self . _quit :
        break
    ret = self . method ( )
    if recurring is None :
        timeout = ret
    elif not recurring :
        break


signature:
def _run_io_threads ( self , handler ) :
    """Start threads for an IOHandler ."""

implementation:
reader = ReadingThread ( self . settings , handler , daemon = self . daemon , exc_queue = self . exc_queue )
writter = WrittingThread ( self . settings , handler , daemon = self . daemon , exc_queue = self . exc_queue )
self . io_threads += [ reader , writter ]
reader . start ( )
writter . start ( )


signature:
def _remove_io_handler ( self , handler ) :
    """Remove an IOHandler from the pool ."""

implementation:
if handler not in self . io_handlers :
    return
self . io_handlers . remove ( handler )
for thread in self . io_threads :
    if thread . io_handler is handler :
        thread . stop ( )


signature:
def _add_timeout_handler ( self , handler ) :
    """Add a TimeoutHandler to the pool ."""

implementation:
self . timeout_handlers . append ( handler )
if self . event_thread is None :
    return
self . _run_timeout_threads ( handler )


signature:
def _run_timeout_threads ( self , handler ) :
    """Start threads for a TimeoutHandler ."""

implementation:
# pylint : disable - msg = W0212
for dummy , method in inspect . getmembers ( handler , callable ) :
    if not hasattr ( method , "_pyxmpp_timeout" ) :
        continue
    thread = TimeoutThread ( method , daemon = self . daemon , exc_queue = self . exc_queue )
    self . timeout_threads . append ( thread )
    thread . start ( )


signature:
def _remove_timeout_handler ( self , handler ) :
    """Remove a TimeoutHandler from the pool ."""

implementation:
if handler not in self . timeout_handlers :
    return
self . timeout_handlers . remove ( handler )
for thread in self . timeout_threads :
    if thread . method . im_self is handler :
        thread . stop ( )


signature:
def start ( self , daemon = False ) :
    """Start the threads ."""

implementation:
self . daemon = daemon
self . io_threads = [ ]
self . event_thread = EventDispatcherThread ( self . event_dispatcher , daemon = daemon , exc_queue = self . exc_queue )
self . event_thread . start ( )
for handler in self . io_handlers :
    self . _run_io_threads ( handler )
for handler in self . timeout_handlers :
    self . _run_timeout_threads ( handler )


signature:
def stop ( self , join = False , timeout = None ) :
    """Stop the threads .
 : Parameters :
 - ` join ` : join the threads ( wait until they exit )
 - ` timeout ` : maximum time ( in seconds ) to wait when ` join ` is
 ` True ` ) . No limit when ` timeout ` is ` None ` ."""

implementation:
logger . debug ( "Closing the io handlers..." )
for handler in self . io_handlers :
    handler . close ( )
if self . event_thread and self . event_thread . is_alive ( ) :
    logger . debug ( "Sending the QUIT signal" )
    self . event_queue . put ( QUIT )
logger . debug ( "  sent" )
threads = self . io_threads + self . timeout_threads
for thread in threads :
    logger . debug ( "Stopping thread: {0!r}" . format ( thread ) )
    thread . stop ( )
if not join :
    return
if self . event_thread :
    threads . append ( self . event_thread )
if timeout is None :
    for thread in threads :
        thread . join ( )
else :
    timeout1 = ( timeout * 0.01 ) / len ( threads )
    threads_left = [ ]
    for thread in threads :
        logger . debug ( "Quick-joining thread {0!r}..." . format ( thread ) )
        thread . join ( timeout1 )
        if thread . is_alive ( ) :
            logger . debug ( "  thread still alive" . format ( thread ) )
            threads_left . append ( thread )
    if threads_left :
        timeout2 = ( timeout * 0.99 ) / len ( threads_left )
        for thread in threads_left :
            logger . debug ( "Joining thread {0!r}..." . format ( thread ) )
            thread . join ( timeout2 )
self . io_threads = [ ]
self . event_thread = None


signature:
def loop_iteration ( self , timeout = 0.1 ) :
    """Wait up to ` timeout ` seconds , raise any exception from the
 threads ."""

implementation:
try :
    exc_info = self . exc_queue . get ( True , timeout ) [ 1 ]
except Queue . Empty :
    return
exc_type , exc_value , ext_stack = exc_info
raise exc_type , exc_value , ext_stack


signature:
def _reset ( self ) :
    """Reset the ` LegacyClientStream ` object state , making the object ready
 to handle new connections ."""

implementation:
ClientStream . _reset ( self )
self . available_auth_methods = None
self . auth_stanza = None
self . registration_callback = None


signature:
def _post_connect ( self ) :
    """Initialize authentication when the connection is established
 and we are the initiator ."""

implementation:
if not self . initiator :
    if "plain" in self . auth_methods or "digest" in self . auth_methods :
        self . set_iq_get_handler ( "query" , "jabber:iq:auth" , self . auth_in_stage1 )
        self . set_iq_set_handler ( "query" , "jabber:iq:auth" , self . auth_in_stage2 )
elif self . registration_callback :
    iq = Iq ( stanza_type = "get" )
    iq . set_content ( Register ( ) )
    self . set_response_handlers ( iq , self . registration_form_received , self . registration_error )
    self . send ( iq )
    return
ClientStream . _post_connect ( self )


signature:
def _post_auth ( self ) :
    """Unregister legacy authentication handlers after successfull
 authentication ."""

implementation:
ClientStream . _post_auth ( self )
if not self . initiator :
    self . unset_iq_get_handler ( "query" , "jabber:iq:auth" )
    self . unset_iq_set_handler ( "query" , "jabber:iq:auth" )


signature:
def _try_auth ( self ) :
    """Try to authenticate using the first one of allowed authentication
 methods left .
 [ client only ]"""

implementation:
if self . authenticated :
    self . __logger . debug ( "try_auth: already authenticated" )
    return
self . __logger . debug ( "trying auth: %r" % ( self . _auth_methods_left , ) )
if not self . _auth_methods_left :
    raise LegacyAuthenticationError ( "No allowed authentication methods available" )
method = self . _auth_methods_left [ 0 ]
if method . startswith ( "sasl:" ) :
    return ClientStream . _try_auth ( self )
elif method not in ( "plain" , "digest" ) :
    self . _auth_methods_left . pop ( 0 )
    self . __logger . debug ( "Skipping unknown auth method: %s" % method )
    return self . _try_auth ( )
elif self . available_auth_methods is not None :
    if method in self . available_auth_methods :
        self . _auth_methods_left . pop ( 0 )
        self . auth_method_used = method
        if method == "digest" :
            self . _digest_auth_stage2 ( self . auth_stanza )
        else :
            self . _plain_auth_stage2 ( self . auth_stanza )
        self . auth_stanza = None
        return
    else :
        self . __logger . debug ( "Skipping unavailable auth method: %s" % method )
else :
    self . _auth_stage1 ( )


signature:
def auth_in_stage1 ( self , stanza ) :
    """Handle the first stage ( < iq type = ' get ' / > ) of legacy ( " plain " or
 " digest " ) authentication .
 [ server only ]"""

implementation:
self . lock . acquire ( )
try :
    if "plain" not in self . auth_methods and "digest" not in self . auth_methods :
        iq = stanza . make_error_response ( "not-allowed" )
        self . send ( iq )
        return
    iq = stanza . make_result_response ( )
    q = iq . new_query ( "jabber:iq:auth" )
    q . newChild ( None , "username" , None )
    q . newChild ( None , "resource" , None )
    if "plain" in self . auth_methods :
        q . newChild ( None , "password" , None )
    if "digest" in self . auth_methods :
        q . newChild ( None , "digest" , None )
    self . send ( iq )
    iq . free ( )
finally :
    self . lock . release ( )


signature:
def auth_in_stage2 ( self , stanza ) :
    """Handle the second stage ( < iq type = ' set ' / > ) of legacy ( " plain " or
 " digest " ) authentication .
 [ server only ]"""

implementation:
self . lock . acquire ( )
try :
    if "plain" not in self . auth_methods and "digest" not in self . auth_methods :
        iq = stanza . make_error_response ( "not-allowed" )
        self . send ( iq )
        return
    username = stanza . xpath_eval ( "a:query/a:username" , { "a" : "jabber:iq:auth" } )
    if username :
        username = from_utf8 ( username [ 0 ] . getContent ( ) )
    resource = stanza . xpath_eval ( "a:query/a:resource" , { "a" : "jabber:iq:auth" } )
    if resource :
        resource = from_utf8 ( resource [ 0 ] . getContent ( ) )
    if not username or not resource :
        self . __logger . debug ( "No username or resource found in auth request" )
        iq = stanza . make_error_response ( "bad-request" )
        self . send ( iq )
        return
    if stanza . xpath_eval ( "a:query/a:password" , { "a" : "jabber:iq:auth" } ) :
        if "plain" not in self . auth_methods :
            iq = stanza . make_error_response ( "not-allowed" )
            self . send ( iq )
            return
        else :
            return self . _plain_auth_in_stage2 ( username , resource , stanza )
    if stanza . xpath_eval ( "a:query/a:digest" , { "a" : "jabber:iq:auth" } ) :
        if "plain" not in self . auth_methods :
            iq = stanza . make_error_response ( "not-allowed" )
            self . send ( iq )
            return
        else :
            return self . _digest_auth_in_stage2 ( username , resource , stanza )
finally :
    self . lock . release ( )


signature:
def _auth_stage1 ( self ) :
    """Do the first stage ( < iq type = ' get ' / > ) of legacy ( " plain " or
 " digest " ) authentication .
 [ client only ]"""

implementation:
iq = Iq ( stanza_type = "get" )
q = iq . new_query ( "jabber:iq:auth" )
q . newTextChild ( None , "username" , to_utf8 ( self . my_jid . node ) )
q . newTextChild ( None , "resource" , to_utf8 ( self . my_jid . resource ) )
self . send ( iq )
self . set_response_handlers ( iq , self . auth_stage2 , self . auth_error , self . auth_timeout , timeout = 60 )
iq . free ( )


signature:
def auth_timeout ( self ) :
    """Handle legacy authentication timeout .
 [ client only ]"""

implementation:
self . lock . acquire ( )
try :
    self . __logger . debug ( "Timeout while waiting for jabber:iq:auth result" )
    if self . _auth_methods_left :
        self . _auth_methods_left . pop ( 0 )
finally :
    self . lock . release ( )


signature:
def auth_error ( self , stanza ) :
    """Handle legacy authentication error .
 [ client only ]"""

implementation:
self . lock . acquire ( )
try :
    err = stanza . get_error ( )
    ae = err . xpath_eval ( "e:*" , { "e" : "jabber:iq:auth:error" } )
    if ae :
        ae = ae [ 0 ] . name
    else :
        ae = err . get_condition ( ) . name
    raise LegacyAuthenticationError ( "Authentication error condition: %s" % ( ae , ) )
finally :
    self . lock . release ( )


signature:
def auth_stage2 ( self , stanza ) :
    """Handle the first stage authentication response ( result of the < iq
 type = " get " / > ) .
 [ client only ]"""

implementation:
self . lock . acquire ( )
try :
    self . __logger . debug ( "Procesing auth response..." )
    self . available_auth_methods = [ ]
    if ( stanza . xpath_eval ( "a:query/a:digest" , { "a" : "jabber:iq:auth" } ) and self . stream_id ) :
        self . available_auth_methods . append ( "digest" )
    if ( stanza . xpath_eval ( "a:query/a:password" , { "a" : "jabber:iq:auth" } ) ) :
        self . available_auth_methods . append ( "plain" )
    self . auth_stanza = stanza . copy ( )
    self . _try_auth ( )
finally :
    self . lock . release ( )


signature:
def _plain_auth_stage2 ( self , _unused ) :
    """Do the second stage ( < iq type = ' set ' / > ) of legacy " plain "
 authentication .
 [ client only ]"""

implementation:
iq = Iq ( stanza_type = "set" )
q = iq . new_query ( "jabber:iq:auth" )
q . newTextChild ( None , "username" , to_utf8 ( self . my_jid . node ) )
q . newTextChild ( None , "resource" , to_utf8 ( self . my_jid . resource ) )
q . newTextChild ( None , "password" , to_utf8 ( self . password ) )
self . send ( iq )
self . set_response_handlers ( iq , self . auth_finish , self . auth_error )
iq . free ( )


signature:
def _plain_auth_in_stage2 ( self , username , _unused , stanza ) :
    """Handle the second stage ( < iq type = ' set ' / > ) of legacy " plain "
 authentication .
 [ server only ]"""

implementation:
password = stanza . xpath_eval ( "a:query/a:password" , { "a" : "jabber:iq:auth" } )
if password :
    password = from_utf8 ( password [ 0 ] . getContent ( ) )
if not password :
    self . __logger . debug ( "No password found in plain auth request" )
    iq = stanza . make_error_response ( "bad-request" )
    self . send ( iq )
    return
if self . check_password ( username , password ) :
    iq = stanza . make_result_response ( )
    self . send ( iq )
    self . peer_authenticated = True
    self . auth_method_used = "plain"
    self . state_change ( "authorized" , self . peer )
    self . _post_auth ( )
else :
    self . __logger . debug ( "Plain auth failed" )
    iq = stanza . make_error_response ( "bad-request" )
    e = iq . get_error ( )
    e . add_custom_condition ( 'jabber:iq:auth:error' , "user-unauthorized" )
    self . send ( iq )


signature:
def _digest_auth_stage2 ( self , _unused ) :
    """Do the second stage ( < iq type = ' set ' / > ) of legacy " digest "
 authentication .
 [ client only ]"""

implementation:
iq = Iq ( stanza_type = "set" )
q = iq . new_query ( "jabber:iq:auth" )
q . newTextChild ( None , "username" , to_utf8 ( self . my_jid . node ) )
q . newTextChild ( None , "resource" , to_utf8 ( self . my_jid . resource ) )
digest = hashlib . sha1 ( to_utf8 ( self . stream_id ) + to_utf8 ( self . password ) ) . hexdigest ( )
q . newTextChild ( None , "digest" , digest )
self . send ( iq )
self . set_response_handlers ( iq , self . auth_finish , self . auth_error )
iq . free ( )


signature:
def _digest_auth_in_stage2 ( self , username , _unused , stanza ) :
    """Handle the second stage ( < iq type = ' set ' / > ) of legacy " digest "
 authentication .
 [ server only ]"""

implementation:
digest = stanza . xpath_eval ( "a:query/a:digest" , { "a" : "jabber:iq:auth" } )
if digest :
    digest = digest [ 0 ] . getContent ( )
if not digest :
    self . __logger . debug ( "No digest found in digest auth request" )
    iq = stanza . make_error_response ( "bad-request" )
    self . send ( iq )
    return
password , pwformat = self . get_password ( username )
if not password or pwformat != "plain" :
    iq = stanza . make_error_response ( "bad-request" )
    e = iq . get_error ( )
    e . add_custom_condition ( 'jabber:iq:auth:error' , "user-unauthorized" )
    self . send ( iq )
    return
mydigest = hashlib . sha1 ( to_utf8 ( self . stream_id ) + to_utf8 ( password ) ) . hexdigest ( )
if mydigest == digest :
    iq = stanza . make_result_response ( )
    self . send ( iq )
    self . peer_authenticated = True
    self . auth_method_used = "digest"
    self . state_change ( "authorized" , self . peer )
    self . _post_auth ( )
else :
    self . __logger . debug ( "Digest auth failed: %r != %r" % ( digest , mydigest ) )
    iq = stanza . make_error_response ( "bad-request" )
    e = iq . get_error ( )
    e . add_custom_condition ( 'jabber:iq:auth:error' , "user-unauthorized" )
    self . send ( iq )


signature:
def auth_finish ( self , _unused ) :
    """Handle success of the legacy authentication ."""

implementation:
self . lock . acquire ( )
try :
    self . __logger . debug ( "Authenticated" )
    self . authenticated = True
    self . state_change ( "authorized" , self . my_jid )
    self . _post_auth ( )
finally :
    self . lock . release ( )


signature:
def registration_error ( self , stanza ) :
    """Handle in - band registration error .
 [ client only ]
 : Parameters :
 - ` stanza ` : the error stanza received or ` None ` on timeout .
 : Types :
 - ` stanza ` : ` pyxmpp . stanza . Stanza `"""

implementation:
self . lock . acquire ( )
try :
    err = stanza . get_error ( )
    ae = err . xpath_eval ( "e:*" , { "e" : "jabber:iq:auth:error" } )
    if ae :
        ae = ae [ 0 ] . name
    else :
        ae = err . get_condition ( ) . name
    raise RegistrationError ( "Authentication error condition: %s" % ( ae , ) )
finally :
    self . lock . release ( )


signature:
def registration_form_received ( self , stanza ) :
    """Handle registration form received .
 [ client only ]
 Call self . registration _ callback with the registration form received
 as the argument . Use the value returned by the callback will be a
 filled - in form .
 : Parameters :
 - ` stanza ` : the stanza received .
 : Types :
 - ` stanza ` : ` pyxmpp . iq . Iq `"""

implementation:
self . lock . acquire ( )
try :
    self . __register = Register ( stanza . get_query ( ) )
    self . registration_callback ( stanza , self . __register . get_form ( ) )
finally :
    self . lock . release ( )


signature:
def submit_registration_form ( self , form ) :
    """Submit a registration form .
 [ client only ]
 : Parameters :
 - ` form ` : the filled - in form . When form is ` None ` or its type is
 " cancel " the registration is to be canceled .
 : Types :
 - ` form ` : ` pyxmpp . jabber . dataforms . Form `"""

implementation:
self . lock . acquire ( )
try :
    if form and form . type != "cancel" :
        self . registration_form = form
        iq = Iq ( stanza_type = "set" )
        iq . set_content ( self . __register . submit_form ( form ) )
        self . set_response_handlers ( iq , self . registration_success , self . registration_error )
        self . send ( iq )
    else :
        self . __register = None
finally :
    self . lock . release ( )


signature:
def registration_success ( self , stanza ) :
    """Handle registration success .
 [ client only ]
 Clean up registration stuff , change state to " registered " and initialize
 authentication .
 : Parameters :
 - ` stanza ` : the stanza received .
 : Types :
 - ` stanza ` : ` pyxmpp . iq . Iq `"""

implementation:
_unused = stanza
self . lock . acquire ( )
try :
    self . state_change ( "registered" , self . registration_form )
    if ( 'FORM_TYPE' in self . registration_form and self . registration_form [ 'FORM_TYPE' ] . value == 'jabber:iq:register' ) :
        if 'username' in self . registration_form :
            self . my_jid = JID ( self . registration_form [ 'username' ] . value , self . my_jid . domain , self . my_jid . resource )
        if 'password' in self . registration_form :
            self . password = self . registration_form [ 'password' ] . value
    self . registration_callback = None
    self . _post_connect ( )
finally :
    self . lock . release ( )


signature:
def _add_io_handler ( self , handler ) :
    """Add an I / O handler to the loop ."""

implementation:
logger . debug ( 'adding io handler: %r' , handler )
self . _unprepared_handlers [ handler ] = None
self . _configure_io_handler ( handler )


signature:
def _configure_io_handler ( self , handler ) :
    """Register an io - handler at the polling object ."""

implementation:
if self . check_events ( ) :
    return
if handler in self . _unprepared_handlers :
    old_fileno = self . _unprepared_handlers [ handler ]
    prepared = self . _prepare_io_handler ( handler )
else :
    old_fileno = None
    prepared = True
fileno = handler . fileno ( )
if old_fileno is not None and fileno != old_fileno :
    del self . _handlers [ old_fileno ]
    # remove _ handler won ' t raise something like KeyError if the fd
    # isn ' t registered ; it will just print a debug log .
    self . io_loop . remove_handler ( old_fileno )
if not prepared :
    self . _unprepared_handlers [ handler ] = fileno
if not fileno :
    return
update = fileno in self . _handlers
events = ioloop . IOLoop . NONE
if handler . is_readable ( ) :
    logger . debug ( " {0!r} readable" . format ( handler ) )
    events |= ioloop . IOLoop . READ
if handler . is_writable ( ) :
    logger . debug ( " {0!r} writable" . format ( handler ) )
    events |= ioloop . IOLoop . WRITE
if self . _handlers . get ( fileno , None ) == events :
    return
self . _handlers [ fileno ] = events
if events :
    logger . debug ( " registering {0!r} handler fileno {1} for" " events {2}" . format ( handler , fileno , events ) )
    if update :
        self . io_loop . update_handler ( fileno , events )
    else :
        self . io_loop . add_handler ( fileno , partial ( self . _handle_event , handler ) , events )


signature:
def _prepare_io_handler ( self , handler ) :
    """Call the ` interfaces . IOHandler . prepare ` method and
 remove the handler from unprepared handler list when done ."""

implementation:
logger . debug ( " preparing handler: {0!r}" . format ( handler ) )
ret = handler . prepare ( )
logger . debug ( "   prepare result: {0!r}" . format ( ret ) )
if isinstance ( ret , HandlerReady ) :
    del self . _unprepared_handlers [ handler ]
    prepared = True
elif isinstance ( ret , PrepareAgain ) :
    if ret . timeout is not None :
        now = time . time ( )
        self . io_loop . add_timeout ( now + ret . timeout , partial ( self . _configure_io_handler , handler ) )
    else :
        self . io_loop . add_callback ( partial ( self . _configure_io_handler , handler ) )
    prepared = False
else :
    raise TypeError ( "Unexpected result type from prepare()" )
return prepared


signature:
def _remove_io_handler ( self , handler ) :
    """Remove an i / o - handler ."""

implementation:
if handler in self . _unprepared_handlers :
    old_fileno = self . _unprepared_handlers [ handler ]
    del self . _unprepared_handlers [ handler ]
else :
    old_fileno = handler . fileno ( )
if old_fileno is not None :
    del self . _handlers [ old_fileno ]
    self . io_loop . remove_handler ( handler . fileno ( ) )


signature:
def _handle_event ( self , handler , fd , event ) :
    """handle I / O events"""

implementation:
# pylint : disable = C0103
logger . debug ( '_handle_event: %r, %r, %r' , handler , fd , event )
if event & ioloop . IOLoop . ERROR :
    handler . handle_hup ( )
if event & ioloop . IOLoop . READ :
    handler . handle_read ( )
if event & ioloop . IOLoop . WRITE :
    handler . handle_write ( )
self . _configure_io_handler ( handler )


signature:
def _os_name_factory ( settings ) :
    """Factory for the : r : ` software _ os setting ` default ."""

implementation:
# pylint : disable - msg = W0613 , W0142
return u"{0} {1} {2}" . format ( platform . system ( ) , platform . release ( ) , platform . machine ( ) )


signature:
def initiate ( self , transport , to = None ) :
    """Initiate an XMPP connection over the ` transport ` .
 : Parameters :
 - ` transport ` : an XMPP transport instance
 - ` to ` : peer name"""

implementation:
with self . lock :
    self . initiator = True
    self . transport = transport
    transport . set_target ( self )
    if to :
        self . peer = JID ( to )
    else :
        self . peer = None
    if transport . is_connected ( ) :
        self . _initiate ( )


signature:
def receive ( self , transport , myname ) :
    """Receive an XMPP connection over the ` transport ` .
 : Parameters :
 - ` transport ` : an XMPP transport instance
 - ` myname ` : local stream endpoint name ."""

implementation:
with self . lock :
    self . transport = transport
    transport . set_target ( self )
    self . me = JID ( myname )
    self . initiator = False
    self . _setup_stream_element_handlers ( )


signature:
def _setup_stream_element_handlers ( self ) :
    """Set up stream element handlers .
 Scans the ` handlers ` list for ` StreamFeatureHandler `
 instances and updates ` _ element _ handlers ` mapping with their
 methods decorated with @ ` stream _ element _ handler `"""

implementation:
# pylint : disable - msg = W0212
if self . initiator :
    mode = "initiator"
else :
    mode = "receiver"
self . _element_handlers = { }
for handler in self . handlers :
    if not isinstance ( handler , StreamFeatureHandler ) :
        continue
    for _unused , meth in inspect . getmembers ( handler , callable ) :
        if not hasattr ( meth , "_pyxmpp_stream_element_handled" ) :
            continue
        element_handled = meth . _pyxmpp_stream_element_handled
        if element_handled in self . _element_handlers : # use only the first matching handler
            continue
        if meth . _pyxmpp_usage_restriction in ( None , mode ) :
            self . _element_handlers [ element_handled ] = meth


signature:
def event ( self , event ) : # pylint : disable - msg = R0201
    """Handle a stream event .
 Called when connection state is changed .
 Should not be called with self . lock acquired !"""

implementation:
event . stream = self
logger . debug ( u"Stream event: {0}" . format ( event ) )
self . settings [ "event_queue" ] . put ( event )
return False


signature:
def transport_connected ( self ) :
    """Called when transport has been connected .
 Send the stream head if initiator ."""

implementation:
with self . lock :
    if self . initiator :
        if self . _output_state is None :
            self . _initiate ( )


signature:
def stream_start ( self , element ) :
    """Process < stream : stream > ( stream start ) tag received from peer .
 ` lock ` is acquired when this method is called .
 : Parameters :
 - ` element ` : root element ( empty ) created by the parser"""

implementation:
with self . lock :
    logger . debug ( "input document: " + element_to_unicode ( element ) )
    if not element . tag . startswith ( STREAM_QNP ) :
        self . _send_stream_error ( "invalid-namespace" )
        raise FatalStreamError ( "Bad stream namespace" )
    if element . tag != STREAM_ROOT_TAG :
        self . _send_stream_error ( "bad-format" )
        raise FatalStreamError ( "Bad root element" )
    if self . _input_state == "restart" :
        event = StreamRestartedEvent ( self . peer )
    else :
        event = StreamConnectedEvent ( self . peer )
    self . _input_state = "open"
    version = element . get ( "version" )
    if version :
        try :
            major , minor = version . split ( "." , 1 )
            major , minor = int ( major ) , int ( minor )
        except ValueError :
            self . _send_stream_error ( "unsupported-version" )
            raise FatalStreamError ( "Unsupported protocol version." )
        self . version = ( major , minor )
    else :
        self . version = ( 0 , 9 )
    if self . version [ 0 ] != 1 and self . version != ( 0 , 9 ) :
        self . _send_stream_error ( "unsupported-version" )
        raise FatalStreamError ( "Unsupported protocol version." )
    peer_lang = element . get ( XML_LANG_QNAME )
    self . peer_language = peer_lang
    if not self . initiator :
        lang = None
        languages = self . settings [ "languages" ]
        while peer_lang :
            if peer_lang in languages :
                lang = peer_lang
                break
            match = LANG_SPLIT_RE . match ( peer_lang )
            if not match :
                break
            peer_lang = match . group ( 0 )
        if lang :
            self . language = lang
    if self . initiator :
        self . stream_id = element . get ( "id" )
        peer = element . get ( "from" )
        if peer :
            peer = JID ( peer )
        if self . peer :
            if peer and peer != self . peer :
                logger . debug ( "peer hostname mismatch: {0!r} != {1!r}" . format ( peer , self . peer ) )
        self . peer = peer
    else :
        to = element . get ( "to" )
        if to :
            to = self . check_to ( to )
            if not to :
                self . _send_stream_error ( "host-unknown" )
                raise FatalStreamError ( 'Bad "to"' )
            self . me = JID ( to )
        peer = element . get ( "from" )
        if peer :
            peer = JID ( peer )
        self . _send_stream_start ( self . generate_id ( ) , stream_to = peer )
        self . _send_stream_features ( )
    self . event ( event )


signature:
def stream_end ( self ) :
    """Process < / stream : stream > ( stream end ) tag received from peer ."""

implementation:
logger . debug ( "Stream ended" )
with self . lock :
    self . _input_state = "closed"
    self . transport . disconnect ( )
    self . _output_state = "closed"


signature:
def _send_stream_start ( self , stream_id = None , stream_to = None ) :
    """Send stream start tag ."""

implementation:
if self . _output_state in ( "open" , "closed" ) :
    raise StreamError ( "Stream start already sent" )
if not self . language :
    self . language = self . settings [ "language" ]
if stream_to :
    stream_to = unicode ( stream_to )
elif self . peer and self . initiator :
    stream_to = unicode ( self . peer )
stream_from = None
if self . me and ( self . tls_established or not self . initiator ) :
    stream_from = unicode ( self . me )
if stream_id :
    self . stream_id = stream_id
else :
    self . stream_id = None
self . transport . send_stream_head ( self . stanza_namespace , stream_from , stream_to , self . stream_id , language = self . language )
self . _output_state = "open"


signature:
def _send_stream_error ( self , condition ) :
    """Same as ` send _ stream _ error ` , but expects ` lock ` acquired ."""

implementation:
if self . _output_state is "closed" :
    return
if self . _output_state in ( None , "restart" ) :
    self . _send_stream_start ( )
element = StreamErrorElement ( condition ) . as_xml ( )
self . transport . send_element ( element )
self . transport . disconnect ( )
self . _output_state = "closed"


signature:
def _restart_stream ( self ) :
    """Restart the stream as needed after SASL and StartTLS negotiation ."""

implementation:
self . _input_state = "restart"
self . _output_state = "restart"
self . features = None
self . transport . restart ( )
if self . initiator :
    self . _send_stream_start ( self . stream_id )


signature:
def _make_stream_features ( self ) :
    """Create the < features / > element for the stream .
 [ receving entity only ]
 : returns : new < features / > element
 : returntype : : etree : ` ElementTree . Element `"""

implementation:
features = ElementTree . Element ( FEATURES_TAG )
for handler in self . _stream_feature_handlers :
    handler . make_stream_features ( self , features )
return features


signature:
def _send_stream_features ( self ) :
    """Send stream < features / > .
 [ receiving entity only ]"""

implementation:
self . features = self . _make_stream_features ( )
self . _write_element ( self . features )


signature:
def _send ( self , stanza ) :
    """Same as ` send ` but assume ` lock ` is acquired ."""

implementation:
self . fix_out_stanza ( stanza )
element = stanza . as_xml ( )
self . _write_element ( element )


signature:
def _process_element ( self , element ) :
    """Process first level element of the stream .
 The element may be stream error or features , StartTLS
 request / response , SASL request / response or a stanza .
 : Parameters :
 - ` element ` : XML element
 : Types :
 - ` element ` : : etree : ` ElementTree . Element `"""

implementation:
tag = element . tag
if tag in self . _element_handlers :
    handler = self . _element_handlers [ tag ]
    logger . debug ( "Passing element {0!r} to method {1!r}" . format ( element , handler ) )
    handled = handler ( self , element )
    if handled :
        return
if tag . startswith ( self . _stanza_namespace_p ) :
    stanza = stanza_factory ( element , self , self . language )
    self . uplink_receive ( stanza )
elif tag == ERROR_TAG :
    error = StreamErrorElement ( element )
    self . process_stream_error ( error )
elif tag == FEATURES_TAG :
    logger . debug ( "Got features element: {0}" . format ( serialize ( element ) ) )
    self . _got_features ( element )
else :
    logger . debug ( "Unhandled element: {0}" . format ( serialize ( element ) ) )
    logger . debug ( " known handlers: {0!r}" . format ( self . _element_handlers ) )


signature:
def uplink_receive ( self , stanza ) :
    """Handle stanza received from the stream ."""

implementation:
with self . lock :
    if self . stanza_route :
        self . stanza_route . uplink_receive ( stanza )
    else :
        logger . debug ( u"Stanza dropped (no route): {0!r}" . format ( stanza ) )


signature:
def process_stream_error ( self , error ) :
    """Process stream error element received .
 : Parameters :
 - ` error ` : error received
 : Types :
 - ` error ` : ` StreamErrorElement `"""

implementation:
# pylint : disable - msg = R0201
logger . debug ( "Unhandled stream error: condition: {0} {1!r}" . format ( error . condition_name , error . serialize ( ) ) )


signature:
def _got_features ( self , features ) :
    """Process incoming < stream : features / > element .
 [ initiating entity only ]
 The received features node is available in ` features ` ."""

implementation:
self . features = features
logger . debug ( "got features, passing to event handlers..." )
handled = self . event ( GotFeaturesEvent ( self . features ) )
logger . debug ( "  handled: {0}" . format ( handled ) )
if not handled :
    mandatory_handled = [ ]
    mandatory_not_handled = [ ]
    logger . debug ( "  passing to stream features handlers: {0}" . format ( self . _stream_feature_handlers ) )
    for handler in self . _stream_feature_handlers :
        ret = handler . handle_stream_features ( self , self . features )
        if ret is None :
            continue
        elif isinstance ( ret , StreamFeatureHandled ) :
            if ret . mandatory :
                mandatory_handled . append ( unicode ( ret ) )
                break
            break
        elif isinstance ( ret , StreamFeatureNotHandled ) :
            if ret . mandatory :
                mandatory_not_handled . append ( unicode ( ret ) )
                break
        else :
            raise ValueError ( "Wrong value returned from a stream" " feature handler: {0!r}" . format ( ret ) )
    if mandatory_not_handled and not mandatory_handled :
        self . send_stream_error ( "unsupported-feature" )
        raise FatalStreamError ( u"Unsupported mandatory-to-implement features: " + u" " . join ( mandatory_not_handled ) )


signature:
def set_peer_authenticated ( self , peer , restart_stream = False ) :
    """Mark the other side of the stream authenticated as ` peer `
 : Parameters :
 - ` peer ` : local JID just authenticated
 - ` restart _ stream ` : ` True ` when stream should be restarted ( needed
 after SASL authentication )
 : Types :
 - ` peer ` : ` JID `
 - ` restart _ stream ` : ` bool `"""

implementation:
with self . lock :
    self . peer_authenticated = True
    self . peer = peer
    if restart_stream :
        self . _restart_stream ( )
self . event ( AuthenticatedEvent ( self . peer ) )


signature:
def set_authenticated ( self , me , restart_stream = False ) :
    """Mark stream authenticated as ` me ` .
 : Parameters :
 - ` me ` : local JID just authenticated
 - ` restart _ stream ` : ` True ` when stream should be restarted ( needed
 after SASL authentication )
 : Types :
 - ` me ` : ` JID `
 - ` restart _ stream ` : ` bool `"""

implementation:
with self . lock :
    self . authenticated = True
    self . me = me
    if restart_stream :
        self . _restart_stream ( )
self . event ( AuthenticatedEvent ( self . me ) )


signature:
def auth_properties ( self ) :
    """Authentication properties of the stream .
 Derived from the transport with ' local - jid ' and ' service - type ' added ."""

implementation:
props = dict ( self . settings [ "extra_auth_properties" ] )
if self . transport :
    props . update ( self . transport . auth_properties )
props [ "local-jid" ] = self . me
props [ "service-type" ] = "xmpp"
return props


signature:
def initiate ( self , transport , to = None ) :
    """Initiate an XMPP connection over the ` transport ` .
 : Parameters :
 - ` transport ` : an XMPP transport instance
 - ` to ` : peer name ( defaults to own jid domain part )"""

implementation:
if to is None :
    to = JID ( self . me . domain )
return StreamBase . initiate ( self , transport , to )


signature:
def receive ( self , transport , myname = None ) :
    """Receive an XMPP connection over the ` transport ` .
 : Parameters :
 - ` transport ` : an XMPP transport instance
 - ` myname ` : local stream endpoint name ( defaults to own jid domain
 part ) ."""

implementation:
if myname is None :
    myname = JID ( self . me . domain )
return StreamBase . receive ( self , transport , myname )


signature:
def fix_out_stanza ( self , stanza ) :
    """Fix outgoing stanza .
 On a client clear the sender JID . On a server set the sender
 address to the own JID if the address is not set yet ."""

implementation:
StreamBase . fix_out_stanza ( self , stanza )
if self . initiator :
    if stanza . from_jid :
        stanza . from_jid = None
else :
    if not stanza . from_jid :
        stanza . from_jid = self . me


signature:
def fix_in_stanza ( self , stanza ) :
    """Fix an incoming stanza .
 Ona server replace the sender address with authorized client JID ."""

implementation:
StreamBase . fix_in_stanza ( self , stanza )
if not self . initiator :
    if stanza . from_jid != self . peer :
        stanza . set_from ( self . peer )


signature:
def loop_iteration ( self , timeout = 60 ) :
    """A loop iteration - check any scheduled events
 and I / O available and run the handlers ."""

implementation:
if self . check_events ( ) :
    return 0
next_timeout , sources_handled = self . _call_timeout_handlers ( )
if self . _quit :
    return sources_handled
if next_timeout is not None :
    timeout = min ( next_timeout , timeout )
readable , writable , next_timeout = self . _prepare_handlers ( )
if next_timeout is not None :
    timeout = min ( next_timeout , timeout )
if not readable and not writable :
    readable , writable , _unused = [ ] , [ ] , None
    time . sleep ( timeout )
else :
    logger . debug ( "select({0!r}, {1!r}, [], {2!r})" . format ( readable , writable , timeout ) )
    readable , writable , _unused = select . select ( readable , writable , [ ] , timeout )
for handler in readable :
    handler . handle_read ( )
    sources_handled += 1
for handler in writable :
    handler . handle_write ( )
    sources_handled += 1
return sources_handled


signature:
def _prepare_handlers ( self ) :
    """Prepare the I / O handlers .
 : Return : ( readable , writable , timeout ) tuple . ' readable ' is the list
 of readable handlers , ' writable ' - the list of writable handlers ,
 ' timeout ' the suggested maximum timeout for this loop iteration or
 ` None `"""

implementation:
timeout = None
readable = [ ]
writable = [ ]
for handler in self . _handlers :
    if handler not in self . _prepared :
        logger . debug ( " preparing handler: {0!r}" . format ( handler ) )
        ret = handler . prepare ( )
        logger . debug ( "   prepare result: {0!r}" . format ( ret ) )
        if isinstance ( ret , HandlerReady ) :
            self . _prepared . add ( handler )
        elif isinstance ( ret , PrepareAgain ) :
            if ret . timeout is not None :
                if timeout is None :
                    timeout = ret . timeout
                else :
                    timeout = min ( timeout , ret . timeout )
        else :
            raise TypeError ( "Unexpected result type from prepare()" )
    if not handler . fileno ( ) :
        logger . debug ( " {0!r}: no fileno" . format ( handler ) )
        continue
    if handler . is_readable ( ) :
        logger . debug ( " {0!r} readable" . format ( handler ) )
        readable . append ( handler )
    if handler . is_writable ( ) :
        logger . debug ( " {0!r} writable" . format ( handler ) )
        writable . append ( handler )
return readable , writable , timeout


signature:
def __from_xml ( self , xmlnode ) :
    """Initialize ` Register ` from an XML node .
 : Parameters :
 - ` xmlnode ` : the jabber : x : register XML element .
 : Types :
 - ` xmlnode ` : ` libxml2 . xmlNode `"""

implementation:
self . __logger . debug ( "Converting jabber:iq:register element from XML" )
if xmlnode . type != "element" :
    raise ValueError ( "XML node is not a jabber:iq:register element (not an element)" )
ns = get_node_ns_uri ( xmlnode )
if ns and ns != REGISTER_NS or xmlnode . name != "query" :
    raise ValueError ( "XML node is not a jabber:iq:register element" )
for element in xml_element_iter ( xmlnode . children ) :
    ns = get_node_ns_uri ( element )
    if ns == DATAFORM_NS and element . name == "x" and not self . form :
        self . form = Form ( element )
    elif ns != REGISTER_NS :
        continue
    name = element . name
    if name == "instructions" and not self . instructions :
        self . instructions = from_utf8 ( element . getContent ( ) )
    elif name == "registered" :
        self . registered = True
    elif name == "remove" :
        self . remove = True
    elif name in legacy_fields and not getattr ( self , name ) :
        value = from_utf8 ( element . getContent ( ) )
        if value is None :
            value = u""
        self . __logger . debug ( u"Setting legacy field %r to %r" % ( name , value ) )
        setattr ( self , name , value )


signature:
def complete_xml_element ( self , xmlnode , doc ) :
    """Complete the XML node with ` self ` content .
 : Parameters :
 - ` xmlnode ` : XML node with the element being built . It has already
 right name and namespace , but no attributes or content .
 - ` doc ` : document to which the element belongs .
 : Types :
 - ` xmlnode ` : ` libxml2 . xmlNode `
 - ` doc ` : ` libxml2 . xmlDoc `"""

implementation:
ns = xmlnode . ns ( )
if self . instructions is not None :
    xmlnode . newTextChild ( ns , "instructions" , to_utf8 ( self . instructions ) )
if self . form :
    self . form . as_xml ( xmlnode , doc )
if self . remove :
    xmlnode . newChild ( ns , "remove" , None )
else :
    if self . registered :
        xmlnode . newChild ( ns , "registered" , None )
    for field in legacy_fields :
        value = getattr ( self , field )
        if value is not None :
            xmlnode . newTextChild ( ns , field , to_utf8 ( value ) )


signature:
def get_form ( self , form_type = "form" ) :
    """Return Data Form for the ` Register ` object .
 Convert legacy fields to a data form if ` self . form ` is ` None ` , return ` self . form ` otherwise .
 : Parameters :
 - ` form _ type ` : If " form " , then a form to fill - in should be
 returned . If " sumbit " , then a form with submitted data .
 : Types :
 - ` form _ type ` : ` unicode `
 : return : ` self . form ` or a form created from the legacy fields
 : returntype : ` pyxmpp . jabber . dataforms . Form `"""

implementation:
if self . form :
    if self . form . type != form_type :
        raise ValueError ( "Bad form type in the jabber:iq:register element" )
    return self . form
form = Form ( form_type , instructions = self . instructions )
form . add_field ( "FORM_TYPE" , [ u"jabber:iq:register" ] , "hidden" )
for field in legacy_fields :
    field_type , field_label = legacy_fields [ field ]
    value = getattr ( self , field )
    if value is None :
        continue
    if form_type == "form" :
        if not value :
            value = None
        form . add_field ( name = field , field_type = field_type , label = field_label , value = value , required = True )
    else :
        form . add_field ( name = field , value = value )
return form


signature:
def submit_form ( self , form ) :
    """Make ` Register ` object for submitting the registration form .
 Convert form data to legacy fields if ` self . form ` is ` None ` .
 : Parameters :
 - ` form ` : The form to submit . Its type doesn ' t have to be " submit "
 ( a " submit " form will be created here ) , so it could be the form
 obtained from ` get _ form ` just with the data entered .
 : return : new registration element
 : returntype : ` Register `"""

implementation:
result = Register ( )
if self . form :
    result . form = form . make_submit ( )
    return result
if "FORM_TYPE" not in form or "jabber:iq:register" not in form [ "FORM_TYPE" ] . values :
    raise ValueError ( "FORM_TYPE is not jabber:iq:register" )
for field in legacy_fields :
    self . __logger . debug ( u"submitted field %r" % ( field , ) )
    value = getattr ( self , field )
    try :
        form_value = form [ field ] . value
    except KeyError :
        if value :
            raise ValueError ( "Required field with no value!" )
        continue
    setattr ( result , field , form_value )
return result


signature:
def get_delays ( stanza ) :
    """Get jabber : x : delay elements from the stanza .
 : Parameters :
 - ` stanza ` : a , probably delayed , stanza .
 : Types :
 - ` stanza ` : ` pyxmpp . stanza . Stanza `
 : return : list of delay tags sorted by the timestamp .
 : returntype : ` list ` of ` Delay `"""

implementation:
delays = [ ]
n = stanza . xmlnode . children
while n :
    if n . type == "element" and get_node_ns_uri ( n ) == DELAY_NS and n . name == "x" :
        delays . append ( Delay ( n ) )
    n = n . next
delays . sort ( )
return delays


signature:
def from_xml ( self , xmlnode ) :
    """Initialize Delay object from an XML node .
 : Parameters :
 - ` xmlnode ` : the jabber : x : delay XML element .
 : Types :
 - ` xmlnode ` : ` libxml2 . xmlNode `"""

implementation:
if xmlnode . type != "element" :
    raise ValueError ( "XML node is not a jabber:x:delay element (not an element)" )
ns = get_node_ns_uri ( xmlnode )
if ns and ns != DELAY_NS or xmlnode . name != "x" :
    raise ValueError ( "XML node is not a jabber:x:delay element" )
stamp = xmlnode . prop ( "stamp" )
if stamp . endswith ( "Z" ) :
    stamp = stamp [ : - 1 ]
if "-" in stamp :
    stamp = stamp . split ( "-" , 1 ) [ 0 ]
try :
    tm = time . strptime ( stamp , "%Y%m%dT%H:%M:%S" )
except ValueError :
    raise BadRequestProtocolError ( "Bad timestamp" )
tm = tm [ 0 : 8 ] + ( 0 , )
self . timestamp = datetime . datetime . fromtimestamp ( time . mktime ( tm ) )
delay_from = from_utf8 ( xmlnode . prop ( "from" ) )
if delay_from :
    try :
        self . delay_from = JID ( delay_from )
    except JIDError :
        raise JIDMalformedProtocolError ( "Bad JID in the jabber:x:delay 'from' attribute" )
else :
    self . delay_from = None
self . reason = from_utf8 ( xmlnode . getContent ( ) )


signature:
def complete_xml_element ( self , xmlnode , _unused ) :
    """Complete the XML node with ` self ` content .
 Should be overriden in classes derived from ` StanzaPayloadObject ` .
 : Parameters :
 - ` xmlnode ` : XML node with the element being built . It has already
 right name and namespace , but no attributes or content .
 - ` _ unused ` : document to which the element belongs .
 : Types :
 - ` xmlnode ` : ` libxml2 . xmlNode `
 - ` _ unused ` : ` libxml2 . xmlDoc `"""

implementation:
tm = self . timestamp . strftime ( "%Y%m%dT%H:%M:%S" )
xmlnode . setProp ( "stamp" , tm )
if self . delay_from :
    xmlnode . setProp ( "from" , self . delay_from . as_utf8 ( ) )
if self . reason :
    xmlnode . setContent ( to_utf8 ( self . reason ) )


signature:
def main ( ) :
    """Parse the command - line arguments and run the bot ."""

implementation:
parser = argparse . ArgumentParser ( description = 'XMPP echo bot' , parents = [ XMPPSettings . get_arg_parser ( ) ] )
parser . add_argument ( 'jid' , metavar = 'JID' , help = 'The bot JID' )
parser . add_argument ( '--debug' , action = 'store_const' , dest = 'log_level' , const = logging . DEBUG , default = logging . INFO , help = 'Print debug messages' )
parser . add_argument ( '--quiet' , const = logging . ERROR , action = 'store_const' , dest = 'log_level' , help = 'Print only error messages' )
parser . add_argument ( '--trace' , action = 'store_true' , help = 'Print XML data sent and received' )
args = parser . parse_args ( )
settings = XMPPSettings ( { "software_name" : "Echo Bot" } )
settings . load_arguments ( args )
if settings . get ( "password" ) is None :
    password = getpass ( "{0!r} password: " . format ( args . jid ) )
    if sys . version_info . major < 3 :
        password = password . decode ( "utf-8" )
    settings [ "password" ] = password
if sys . version_info . major < 3 :
    args . jid = args . jid . decode ( "utf-8" )
logging . basicConfig ( level = args . log_level )
if args . trace :
    print "enabling trace"
    handler = logging . StreamHandler ( )
    handler . setLevel ( logging . DEBUG )
    for logger in ( "pyxmpp2.IN" , "pyxmpp2.OUT" ) :
        logger = logging . getLogger ( logger )
        logger . setLevel ( logging . DEBUG )
        logger . addHandler ( handler )
        logger . propagate = False
bot = EchoBot ( JID ( args . jid ) , settings )
try :
    bot . run ( )
except KeyboardInterrupt :
    bot . disconnect ( )


signature:
def handle_message ( self , stanza ) :
    """Echo every non - error ` ` < message / > ` ` stanza .
 Add " Re : " to subject , if any ."""

implementation:
if stanza . subject :
    subject = u"Re: " + stanza . subject
else :
    subject = None
msg = Message ( stanza_type = stanza . stanza_type , from_jid = stanza . to_jid , to_jid = stanza . from_jid , subject = subject , body = stanza . body , thread = stanza . thread )
return msg


signature:
def prepare ( self ) :
    """When connecting start the next connection step and schedule
 next ` prepare ` call , when connected return ` HandlerReady ( ) `"""

implementation:
with self . _lock :
    if self . _socket :
        self . _socket . listen ( SOMAXCONN )
        self . _socket . setblocking ( False )
    return HandlerReady ( )


signature:
def handle_read ( self ) :
    """Accept any incoming connections ."""

implementation:
with self . _lock :
    logger . debug ( "handle_read()" )
    if self . _socket is None :
        return
    while True :
        try :
            sock , address = self . _socket . accept ( )
        except socket . error , err :
            if err . args [ 0 ] in BLOCKING_ERRORS :
                break
            else :
                raise
        logger . debug ( "Accepted connection from: {0!r}" . format ( address ) )
        self . _target ( sock , address )


signature:
def _decode_subelements ( self ) :
    """Decode the stanza subelements ."""

implementation:
for child in self . _element :
    if child . tag == self . _show_tag :
        self . _show = child . text
    elif child . tag == self . _status_tag :
        self . _status = child . text
    elif child . tag == self . _priority_tag :
        try :
            self . _priority = int ( child . text . strip ( ) )
            if self . _priority < - 128 or self . _priority > 127 :
                raise ValueError
        except ValueError :
            raise BadRequestProtocolError ( "Presence priority not an integer" )


signature:
def as_xml ( self ) :
    """Return the XML stanza representation .
 Always return an independent copy of the stanza XML representation ,
 which can be freely modified without affecting the stanza .
 : returntype : : etree : ` ElementTree . Element `"""

implementation:
result = Stanza . as_xml ( self )
if self . _show :
    child = ElementTree . SubElement ( result , self . _show_tag )
    child . text = self . _show
if self . _status :
    child = ElementTree . SubElement ( result , self . _status_tag )
    child . text = self . _status
if self . _priority :
    child = ElementTree . SubElement ( result , self . _priority_tag )
    child . text = unicode ( self . _priority )
return result


signature:
def copy ( self ) :
    """Create a deep copy of the stanza .
 : returntype : ` Presence `"""

implementation:
result = Presence ( None , self . from_jid , self . to_jid , self . stanza_type , self . stanza_id , self . error , self . _return_path ( ) , self . _show , self . _status , self . _priority )
if self . _payload is None :
    self . decode_payload ( )
for payload in self . _payload :
    result . add_payload ( payload . copy ( ) )
return result


signature:
def make_accept_response ( self ) :
    """Create " accept " response for the " subscribe " / " subscribed " /
 " unsubscribe " / " unsubscribed " presence stanza .
 : return : new stanza .
 : returntype : ` Presence `"""

implementation:
if self . stanza_type not in ( "subscribe" , "subscribed" , "unsubscribe" , "unsubscribed" ) :
    raise ValueError ( "Results may only be generated for 'subscribe'," "'subscribed','unsubscribe' or 'unsubscribed' presence" )
stanza = Presence ( stanza_type = ACCEPT_RESPONSES [ self . stanza_type ] , from_jid = self . to_jid , to_jid = self . from_jid , stanza_id = self . stanza_id )
return stanza


signature:
def make_deny_response ( self ) :
    """Create " deny " response for the " subscribe " / " subscribed " /
 " unsubscribe " / " unsubscribed " presence stanza .
 : return : new presence stanza .
 : returntype : ` Presence `"""

implementation:
if self . stanza_type not in ( "subscribe" , "subscribed" , "unsubscribe" , "unsubscribed" ) :
    raise ValueError ( "Results may only be generated for 'subscribe'," "'subscribed','unsubscribe' or 'unsubscribed' presence" )
stanza = Presence ( stanza_type = DENY_RESPONSES [ self . stanza_type ] , from_jid = self . to_jid , to_jid = self . from_jid , stanza_id = self . stanza_id )
return stanza


signature:
def make_error_response ( self , cond ) :
    """Create error response for the any non - error presence stanza .
 : Parameters :
 - ` cond ` : error condition name , as defined in XMPP specification .
 : Types :
 - ` cond ` : ` unicode `
 : return : new presence stanza .
 : returntype : ` Presence `"""

implementation:
if self . stanza_type == "error" :
    raise ValueError ( "Errors may not be generated in response" " to errors" )
stanza = Presence ( stanza_type = "error" , from_jid = self . from_jid , to_jid = self . to_jid , stanza_id = self . stanza_id , status = self . _status , show = self . _show , priority = self . _priority , error_cond = cond )
if self . _payload is None :
    self . decode_payload ( )
for payload in self . _payload :
    stanza . add_payload ( payload )
return stanza


signature:
def activate ( self ) :
    """Activate an plan in a CREATED state ."""

implementation:
obj = self . find_paypal_object ( )
if obj . state == enums . BillingPlanState . CREATED :
    success = obj . activate ( )
    if not success :
        raise PaypalApiError ( "Failed to activate plan: %r" % ( obj . error ) )
# Resync the updated data to the database
self . get_or_update_from_api_data ( obj , always_sync = True )
return obj


signature:
def execute ( self ) :
    """Execute the PreparedBillingAgreement by creating and executing a
 matching BillingAgreement ."""

implementation:
# Save the execution time first .
# If execute ( ) fails , executed _ at will be set , with no executed _ agreement set .
self . executed_at = now ( )
self . save ( )
with transaction . atomic ( ) :
    ret = BillingAgreement . execute ( self . id )
    ret . user = self . user
    ret . save ( )
    self . executed_agreement = ret
    self . save ( )
return ret


signature:
def webhook_handler ( * event_types ) :
    """Decorator that registers a function as a webhook handler .
 Usage examples :
 > > > # Hook a single event
 > > > @ webhook _ handler ( " payment . sale . completed " )
 > > > def on _ payment _ received ( event ) :
 > > > payment = event . get _ resource ( )
 > > > print ( " Received payment : " , payment )
 > > > # Multiple events supported
 > > > @ webhook _ handler ( " billing . subscription . suspended " , " billing . subscription . cancelled " )
 > > > def on _ subscription _ stop ( event ) :
 > > > subscription = event . get _ resource ( )
 > > > print ( " Stopping subscription : " , subscription )
 > > > # Using a wildcard works as well
 > > > @ webhook _ handler ( " billing . subscription . * " )
 > > > def on _ subscription _ update ( event ) :
 > > > subscription = event . get _ resource ( )
 > > > print ( " Updated subscription : " , subscription )"""

implementation:
# First expand all wildcards and verify the event types are valid
event_types_to_register = set ( )
for event_type in event_types : # Always convert to lowercase
    event_type = event_type . lower ( )
    if "*" in event_type : # expand it
        for t in WEBHOOK_EVENT_TYPES :
            if fnmatch ( t , event_type ) :
                event_types_to_register . add ( t )
    elif event_type not in WEBHOOK_EVENT_TYPES :
        raise ValueError ( "Unknown webhook event: %r" % ( event_type ) )
    else :
        event_types_to_register . add ( event_type )
# Now register them
def decorator ( func ) :
    for event_type in event_types_to_register :
        WEBHOOK_SIGNALS [ event_type ] . connect ( func )
    return func
return decorator


signature:
def from_request ( cls , request , webhook_id = PAYPAL_WEBHOOK_ID ) :
    """Create , validate and process a WebhookEventTrigger given a Django
 request object .
 The webhook _ id parameter expects the ID of the Webhook that was
 triggered ( defaults to settings . PAYPAL _ WEBHOOK _ ID ) . This is required
 for Webhook verification .
 The process is three - fold :
 1 . Create a WebhookEventTrigger object from a Django request .
 2 . Verify the WebhookEventTrigger as a Paypal webhook using the SDK .
 3 . If valid , process it into a WebhookEvent object ( and child resource ) ."""

implementation:
headers = fix_django_headers ( request . META )
assert headers
try :
    body = request . body . decode ( request . encoding or "utf-8" )
except Exception :
    body = "(error decoding body)"
ip = request . META [ "REMOTE_ADDR" ]
obj = cls . objects . create ( headers = headers , body = body , remote_ip = ip )
try :
    obj . valid = obj . verify ( PAYPAL_WEBHOOK_ID )
    if obj . valid : # Process the item ( do not save it , it ' ll get saved below )
        obj . process ( save = False )
except Exception as e :
    max_length = WebhookEventTrigger . _meta . get_field ( "exception" ) . max_length
    obj . exception = str ( e ) [ : max_length ]
    obj . traceback = format_exc ( )
finally :
    obj . save ( )
return obj


signature:
def run ( * args ) :
    """Check and / or create Django migrations .
 If - - check is present in the arguments then migrations are checked only ."""

implementation:
if not settings . configured :
    settings . configure ( ** DEFAULT_SETTINGS )
django . setup ( )
parent = os . path . dirname ( os . path . abspath ( __file__ ) )
sys . path . insert ( 0 , parent )
if "--check" in args :
    check_migrations ( )
else :
    django . core . management . call_command ( "makemigrations" , APP_NAME , * args )


signature:
def check_paypal_api_key ( app_configs = None , ** kwargs ) :
    """Check that the Paypal API keys are configured correctly"""

implementation:
messages = [ ]
mode = getattr ( djpaypal_settings , "PAYPAL_MODE" , None )
if mode not in VALID_MODES :
    msg = "Invalid PAYPAL_MODE specified: {}." . format ( repr ( mode ) )
    hint = "PAYPAL_MODE must be one of {}" . format ( ", " . join ( repr ( k ) for k in VALID_MODES ) )
    messages . append ( checks . Critical ( msg , hint = hint , id = "djpaypal.C001" ) )
for setting in "PAYPAL_CLIENT_ID" , "PAYPAL_CLIENT_SECRET" :
    if not getattr ( djpaypal_settings , setting , None ) :
        msg = "Invalid value specified for {}" . format ( setting )
        hint = "Add PAYPAL_CLIENT_ID and PAYPAL_CLIENT_SECRET to your settings."
        messages . append ( checks . Critical ( msg , hint = hint , id = "djpaypal.C002" ) )
return messages


signature:
async def _create_upstream_applications ( self ) :
    """Create the upstream applications ."""

implementation:
loop = asyncio . get_event_loop ( )
for steam_name , ApplicationsCls in self . applications . items ( ) :
    application = ApplicationsCls ( self . scope )
    upstream_queue = asyncio . Queue ( )
    self . application_streams [ steam_name ] = upstream_queue
    self . application_futures [ steam_name ] = loop . create_task ( application ( upstream_queue . get , partial ( self . dispatch_downstream , steam_name = steam_name ) ) )


signature:
async def send_upstream ( self , message , stream_name = None ) :
    """Send a message upstream to a de - multiplexed application .
 If stream _ name is includes will send just to that upstream steam , if not included will send ot all upstream
 steams ."""

implementation:
if stream_name is None :
    for steam_queue in self . application_streams . values ( ) :
        await steam_queue . put ( message )
    return
steam_queue = self . application_streams . get ( stream_name )
if steam_queue is None :
    raise ValueError ( "Invalid multiplexed frame received (stream not mapped)" )
await steam_queue . put ( message )


signature:
async def dispatch_downstream ( self , message , steam_name ) :
    """Handle a downstream message coming from an upstream steam .
 if there is not handling method set for this method type it will propagate the message further downstream .
 This is called as part of the co - routine of an upstream steam , not the same loop as used for upstream messages
 in the de - multiplexer ."""

implementation:
handler = getattr ( self , get_handler_name ( message ) , None )
if handler :
    await handler ( message , stream_name = steam_name )
else : # if there is not handler then just pass the message further downstream .
    await self . base_send ( message )


signature:
async def receive_json ( self , content , ** kwargs ) :
    """Rout the message down the correct stream ."""

implementation:
# Check the frame looks good
if isinstance ( content , dict ) and "stream" in content and "payload" in content : # Match it to a channel
    steam_name = content [ "stream" ]
    payload = content [ "payload" ]
    # block upstream frames
    if steam_name not in self . applications_accepting_frames :
        raise ValueError ( "Invalid multiplexed frame received (stream not mapped)" )
    # send it on to the application that handles this stream
    await self . send_upstream ( message = { "type" : "websocket.receive" , "text" : await self . encode_json ( payload ) } , stream_name = steam_name )
    return
else :
    raise ValueError ( "Invalid multiplexed **frame received (no channel/payload key)" )


signature:
async def websocket_disconnect ( self , message ) :
    """Handle the disconnect message .
 This is propagated to all upstream applications ."""

implementation:
# set this flag so as to ensure we don ' t send a downstream ` websocket . close ` message due to all
# child applications closing .
self . closing = True
# inform all children
await self . send_upstream ( message )
await super ( ) . websocket_disconnect ( message )


signature:
async def disconnect ( self , code ) :
    """default is to wait for the child applications to close ."""

implementation:
try :
    await asyncio . wait ( self . application_futures . values ( ) , return_when = asyncio . ALL_COMPLETED , timeout = self . application_close_timeout )
except asyncio . TimeoutError :
    pass


signature:
async def websocket_send ( self , message , stream_name ) :
    """Capture downstream websocket . send messages from the upstream applications ."""

implementation:
text = message . get ( "text" )
# todo what to do on binary !
json = await self . decode_json ( text )
data = { "stream" : stream_name , "payload" : json }
await self . send_json ( data )


signature:
async def websocket_accept ( self , message , stream_name ) :
    """Intercept downstream ` websocket . accept ` message and thus allow this upsteam application to accept websocket
 frames ."""

implementation:
is_first = not self . applications_accepting_frames
self . applications_accepting_frames . add ( stream_name )
# accept the connection after the first upstream application accepts .
if is_first :
    await self . accept ( )


signature:
async def websocket_close ( self , message , stream_name ) :
    """Handle downstream ` websocket . close ` message .
 Will disconnect this upstream application from receiving any new frames .
 If there are not more upstream applications accepting messages it will then call ` close ` ."""

implementation:
if stream_name in self . applications_accepting_frames : # remove from set of upsteams steams than can receive new messages
    self . applications_accepting_frames . remove ( stream_name )
# we are already closing due to an upstream websocket . disconnect command
if self . closing :
    return
# if none of the upstream applications are listing we need to close .
if not self . applications_accepting_frames :
    await self . close ( message . get ( "code" ) )