<s> """<STR_LIT>""" <EOL> import logging <EOL> import os <EOL> import random <EOL> import select <EOL> import socket <EOL> import sys <EOL> import time <EOL> from binascii import hexlify <EOL> from contextlib import contextmanager <EOL> from kazoo . exceptions import ( <EOL> AuthFailedError , <EOL> ConnectionDropped , <EOL> EXCEPTIONS , <EOL> SessionExpiredError , <EOL> NoNodeError , <EOL> SaslException <EOL> ) <EOL> from kazoo . handlers . utils import create_pipe <EOL> from kazoo . loggingsupport import BLATHER <EOL> from kazoo . protocol . serialization import ( <EOL> Auth , <EOL> Close , <EOL> Connect , <EOL> Exists , <EOL> GetChildren , <EOL> Ping , <EOL> PingInstance , <EOL> ReplyHeader , <EOL> SASL , <EOL> Transaction , <EOL> Watch , <EOL> int_struct <EOL> ) <EOL> from kazoo . protocol . states import ( <EOL> Callback , <EOL> KeeperState , <EOL> WatchedEvent , <EOL> EVENT_TYPE_MAP , <EOL> ) <EOL> from kazoo . retry import ( <EOL> ForceRetryError , <EOL> RetryFailedError <EOL> ) <EOL> import sasl <EOL> log = logging . getLogger ( __name__ ) <EOL> _SESSION_EXPIRED = object ( ) <EOL> _CONNECTION_DROP = object ( ) <EOL> STOP_CONNECTING = object ( ) <EOL> CREATED_EVENT = <NUM_LIT:1> <EOL> DELETED_EVENT = <NUM_LIT:2> <EOL> CHANGED_EVENT = <NUM_LIT:3> <EOL> CHILD_EVENT = <NUM_LIT:4> <EOL> WATCH_XID = - <NUM_LIT:1> <EOL> PING_XID = - <NUM_LIT:2> <EOL> AUTH_XID = - <NUM_LIT:4> <EOL> CLOSE_RESPONSE = Close . type <EOL> if sys . version_info > ( <NUM_LIT:3> , ) : <EOL> def buffer ( obj , offset = <NUM_LIT:0> ) : <EOL> return memoryview ( obj ) [ offset : ] <EOL> advance_iterator = next <EOL> else : <EOL> def advance_iterator ( it ) : <EOL> return it . next ( ) <EOL> class RWPinger ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hosts , connection_func , socket_handling ) : <EOL> self . hosts = hosts <EOL> self . connection = connection_func <EOL> self . last_attempt = None <EOL> self . socket_handling = socket_handling <EOL> def __iter__ ( self ) : <EOL> if not self . last_attempt : <EOL> self . last_attempt = time . time ( ) <EOL> delay = <NUM_LIT:0.5> <EOL> while True : <EOL> yield self . _next_server ( delay ) <EOL> def _next_server ( self , delay ) : <EOL> jitter = random . randint ( <NUM_LIT:0> , <NUM_LIT:100> ) / <NUM_LIT> <EOL> while time . time ( ) < self . last_attempt + delay + jitter : <EOL> return False <EOL> for host , port in self . hosts : <EOL> log . debug ( "<STR_LIT>" , host , port ) <EOL> self . last_attempt = time . time ( ) <EOL> try : <EOL> with self . socket_handling ( ) : <EOL> sock = self . connection ( ( host , port ) ) <EOL> sock . sendall ( b"<STR_LIT>" ) <EOL> result = sock . recv ( <NUM_LIT> ) <EOL> sock . close ( ) <EOL> if result == b'<STR_LIT>' : <EOL> return ( host , port ) <EOL> else : <EOL> return False <EOL> except ConnectionDropped : <EOL> return False <EOL> while time . time ( ) < self . last_attempt + jitter : <EOL> return False <EOL> delay *= <NUM_LIT:2> <EOL> class RWServerAvailable ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class ConnectionHandler ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , retry_sleeper , logger = None , sasl_server_principal = None ) : <EOL> self . client = client <EOL> self . handler = client . handler <EOL> self . retry_sleeper = retry_sleeper <EOL> self . logger = logger or log <EOL> self . sasl_server_principal = sasl_server_principal <EOL> self . connection_closed = client . handler . event_object ( ) <EOL> self . connection_closed . set ( ) <EOL> self . connection_stopped = client . handler . event_object ( ) <EOL> self . connection_stopped . set ( ) <EOL> self . ping_outstanding = client . handler . event_object ( ) <EOL> self . _read_pipe = None <EOL> self . _write_pipe = None <EOL> self . _socket = None <EOL> self . _xid = None <EOL> self . _rw_server = None <EOL> self . _ro_mode = False <EOL> self . _connection_routine = None <EOL> @ contextmanager <EOL> def _socket_error_handling ( self ) : <EOL> try : <EOL> yield <EOL> except ( socket . error , select . error ) as e : <EOL> err = getattr ( e , '<STR_LIT>' , e ) <EOL> raise ConnectionDropped ( "<STR_LIT>" % ( err , ) ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connection_closed . is_set ( ) : <EOL> self . _read_pipe , self . _write_pipe = create_pipe ( ) <EOL> self . connection_closed . clear ( ) <EOL> if self . _connection_routine : <EOL> raise Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . _connection_routine = self . handler . spawn ( self . zk_loop ) <EOL> def stop ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . connection_stopped . wait ( timeout ) <EOL> if self . _connection_routine : <EOL> self . _connection_routine . join ( ) <EOL> self . _connection_routine = None <EOL> return self . connection_stopped . is_set ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . connection_stopped . is_set ( ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> self . connection_closed . set ( ) <EOL> wp , rp = self . _write_pipe , self . _read_pipe <EOL> self . _write_pipe = self . _read_pipe = None <EOL> if wp is not None : <EOL> os . close ( wp ) <EOL> if rp is not None : <EOL> os . close ( rp ) <EOL> def _server_pinger ( self ) : <EOL> """<STR_LIT>""" <EOL> return RWPinger ( self . client . hosts , self . handler . create_connection , <EOL> self . _socket_error_handling ) <EOL> def _read_header ( self , timeout ) : <EOL> b = self . _read ( <NUM_LIT:4> , timeout ) <EOL> length = int_struct . unpack ( b ) [ <NUM_LIT:0> ] <EOL> b = self . _read ( length , timeout ) <EOL> header , offset = ReplyHeader . deserialize ( b , <NUM_LIT:0> ) <EOL> return header , b , offset <EOL> def _read ( self , length , timeout ) : <EOL> msgparts = [ ] <EOL> remaining = length <EOL> with self . _socket_error_handling ( ) : <EOL> while remaining > <NUM_LIT:0> : <EOL> s = self . handler . select ( [ self . _socket ] , [ ] , [ ] , timeout ) [ <NUM_LIT:0> ] <EOL> if not s : <EOL> raise self . handler . timeout_exception ( "<STR_LIT>" ) <EOL> chunk = self . _socket . recv ( remaining ) <EOL> if chunk == b'<STR_LIT>' : <EOL> raise ConnectionDropped ( '<STR_LIT>' ) <EOL> msgparts . append ( chunk ) <EOL> remaining -= len ( chunk ) <EOL> return b"<STR_LIT>" . join ( msgparts ) <EOL> def _invoke ( self , timeout , request , xid = None ) : <EOL> """<STR_LIT>""" <EOL> self . _submit ( request , timeout , xid ) <EOL> zxid = None <EOL> if xid : <EOL> header , buffer , offset = self . _read_header ( timeout ) <EOL> if header . xid != xid : <EOL> raise RuntimeError ( '<STR_LIT>' , <EOL> xid , header . xid ) <EOL> if header . zxid > <NUM_LIT:0> : <EOL> zxid = header . zxid <EOL> if header . err : <EOL> callback_exception = EXCEPTIONS [ header . err ] ( ) <EOL> self . logger . debug ( <EOL> '<STR_LIT>' , xid , callback_exception ) <EOL> raise callback_exception <EOL> return zxid <EOL> msg = self . _read ( <NUM_LIT:4> , timeout ) <EOL> length = int_struct . unpack ( msg ) [ <NUM_LIT:0> ] <EOL> msg = self . _read ( length , timeout ) <EOL> if hasattr ( request , '<STR_LIT>' ) : <EOL> try : <EOL> obj , _ = request . deserialize ( msg , <NUM_LIT:0> ) <EOL> except Exception : <EOL> self . logger . exception ( "<STR_LIT>" <EOL> "<STR_LIT>" , request ) <EOL> raise ConnectionDropped ( '<STR_LIT>' ) <EOL> self . logger . log ( BLATHER , '<STR_LIT>' , obj ) <EOL> return obj , zxid <EOL> return zxid <EOL> def _submit ( self , request , timeout , xid = None ) : <EOL> """<STR_LIT>""" <EOL> b = bytearray ( ) <EOL> if xid : <EOL> b . extend ( int_struct . pack ( xid ) ) <EOL> if request . type : <EOL> b . extend ( int_struct . pack ( request . type ) ) <EOL> b += request . serialize ( ) <EOL> self . logger . log ( ( BLATHER if isinstance ( request , Ping ) else logging . DEBUG ) , <EOL> "<STR_LIT>" , xid , request ) <EOL> self . _write ( int_struct . pack ( len ( b ) ) + b , timeout ) <EOL> def _write ( self , msg , timeout ) : <EOL> """<STR_LIT>""" <EOL> sent = <NUM_LIT:0> <EOL> msg_length = len ( msg ) <EOL> with self . _socket_error_handling ( ) : <EOL> while sent < msg_length : <EOL> s = self . handler . select ( [ ] , [ self . _socket ] , [ ] , timeout ) [ <NUM_LIT:1> ] <EOL> if not s : <EOL> raise self . handler . timeout_exception ( "<STR_LIT>" ) <EOL> msg_slice = buffer ( msg , sent ) <EOL> bytes_sent = self . _socket . send ( msg_slice ) <EOL> if not bytes_sent : <EOL> raise ConnectionDropped ( '<STR_LIT>' ) <EOL> sent += bytes_sent <EOL> def _read_watch_event ( self , buffer , offset ) : <EOL> client = self . client <EOL> watch , offset = Watch . deserialize ( buffer , offset ) <EOL> path = watch . path <EOL> self . logger . debug ( '<STR_LIT>' , watch ) <EOL> watchers = [ ] <EOL> if watch . type in ( CREATED_EVENT , CHANGED_EVENT ) : <EOL> watchers . extend ( client . _data_watchers . pop ( path , [ ] ) ) <EOL> elif watch . type == DELETED_EVENT : <EOL> watchers . extend ( client . _data_watchers . pop ( path , [ ] ) ) <EOL> watchers . extend ( client . _child_watchers . pop ( path , [ ] ) ) <EOL> elif watch . type == CHILD_EVENT : <EOL> watchers . extend ( client . _child_watchers . pop ( path , [ ] ) ) <EOL> else : <EOL> self . logger . warn ( '<STR_LIT>' , watch . type ) <EOL> return <EOL> path = client . unchroot ( path ) <EOL> ev = WatchedEvent ( EVENT_TYPE_MAP [ watch . type ] , client . _state , path ) <EOL> if client . _stopped . is_set ( ) : <EOL> return <EOL> for watch in watchers : <EOL> client . handler . dispatch_callback ( Callback ( '<STR_LIT>' , watch , ( ev , ) ) ) <EOL> def _read_response ( self , header , buffer , offset ) : <EOL> client = self . client <EOL> request , async_object , xid = client . _pending . popleft ( ) <EOL> if header . zxid and header . zxid > <NUM_LIT:0> : <EOL> client . last_zxid = header . zxid <EOL> if header . xid != xid : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' , xid , header . xid ) <EOL> exists_error = ( header . err == NoNodeError . code and <EOL> request . type == Exists . type ) <EOL> if header . err and not exists_error : <EOL> callback_exception = EXCEPTIONS [ header . err ] ( ) <EOL> self . logger . debug ( <EOL> '<STR_LIT>' , xid , callback_exception ) <EOL> if async_object : <EOL> async_object . set_exception ( callback_exception ) <EOL> elif request and async_object : <EOL> if exists_error : <EOL> async_object . set ( None ) <EOL> else : <EOL> try : <EOL> response = request . deserialize ( buffer , offset ) <EOL> except Exception as exc : <EOL> self . logger . exception ( "<STR_LIT>" <EOL> "<STR_LIT>" , request ) <EOL> async_object . set_exception ( exc ) <EOL> return <EOL> self . logger . debug ( <EOL> '<STR_LIT>' , xid , response ) <EOL> if request . type == Transaction . type : <EOL> response = Transaction . unchroot ( client , response ) <EOL> async_object . set ( response ) <EOL> watcher = getattr ( request , '<STR_LIT>' , None ) <EOL> if not client . _stopped . is_set ( ) and watcher : <EOL> if isinstance ( request , GetChildren ) : <EOL> client . _child_watchers [ request . path ] . add ( watcher ) <EOL> else : <EOL> client . _data_watchers [ request . path ] . add ( watcher ) <EOL> if isinstance ( request , Close ) : <EOL> self . logger . log ( BLATHER , '<STR_LIT>' ) <EOL> return CLOSE_RESPONSE <EOL> def _read_socket ( self , read_timeout ) : <EOL> """<STR_LIT>""" <EOL> client = self . client <EOL> header , buffer , offset = self . _read_header ( read_timeout ) <EOL> if header . xid == PING_XID : <EOL> self . logger . log ( BLATHER , '<STR_LIT>' ) <EOL> self . ping_outstanding . clear ( ) <EOL> elif header . xid == AUTH_XID : <EOL> self . logger . log ( BLATHER , '<STR_LIT>' ) <EOL> request , async_object , xid = client . _pending . popleft ( ) <EOL> if header . err : <EOL> async_object . set_exception ( AuthFailedError ( ) ) <EOL> client . _session_callback ( KeeperState . AUTH_FAILED ) <EOL> else : <EOL> async_object . set ( True ) <EOL> elif header . xid == WATCH_XID : <EOL> self . _read_watch_event ( buffer , offset ) <EOL> else : <EOL> self . logger . log ( BLATHER , '<STR_LIT>' , header ) <EOL> return self . _read_response ( header , buffer , offset ) <EOL> def _send_request ( self , read_timeout , connect_timeout ) : <EOL> """<STR_LIT>""" <EOL> client = self . client <EOL> try : <EOL> request , async_object = client . _queue [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> try : <EOL> os . read ( self . _read_pipe , <NUM_LIT:1> ) <EOL> except OSError : <EOL> pass <EOL> return <EOL> if request is _SESSION_EXPIRED : <EOL> raise SessionExpiredError ( "<STR_LIT>" ) <EOL> if request is _CONNECTION_DROP : <EOL> raise ConnectionDropped ( "<STR_LIT>" ) <EOL> if request . type == Auth . type : <EOL> xid = AUTH_XID <EOL> else : <EOL> self . _xid += <NUM_LIT:1> <EOL> xid = self . _xid <EOL> self . _submit ( request , connect_timeout , xid ) <EOL> client . _queue . popleft ( ) <EOL> os . read ( self . _read_pipe , <NUM_LIT:1> ) <EOL> client . _pending . append ( ( request , async_object , xid ) ) <EOL> def _send_ping ( self , connect_timeout ) : <EOL> self . ping_outstanding . set ( ) <EOL> self . _submit ( PingInstance , connect_timeout , PING_XID ) <EOL> if self . _ro_mode : <EOL> result = advance_iterator ( self . _ro_mode ) <EOL> if result : <EOL> self . _rw_server = result <EOL> raise RWServerAvailable ( ) <EOL> def zk_loop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . logger . log ( BLATHER , '<STR_LIT>' ) <EOL> self . connection_stopped . clear ( ) <EOL> retry = self . retry_sleeper . copy ( ) <EOL> try : <EOL> while not self . client . _stopped . is_set ( ) : <EOL> if retry ( self . _connect_loop , retry ) is STOP_CONNECTING : <EOL> break <EOL> except RetryFailedError : <EOL> self . logger . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> self . connection_stopped . set ( ) <EOL> self . client . _session_callback ( KeeperState . CLOSED ) <EOL> self . logger . log ( BLATHER , '<STR_LIT>' ) <EOL> def _connect_loop ( self , retry ) : <EOL> status = None <EOL> for host , port in self . client . hosts : <EOL> if self . client . _stopped . is_set ( ) : <EOL> status = STOP_CONNECTING <EOL> break <EOL> status = self . _connect_attempt ( host , port , retry ) <EOL> if status is STOP_CONNECTING : <EOL> break <EOL> if status is STOP_CONNECTING : <EOL> return STOP_CONNECTING <EOL> else : <EOL> raise ForceRetryError ( '<STR_LIT>' ) <EOL> def _connect_attempt ( self , host , port , retry ) : <EOL> client = self . client <EOL> TimeoutError = self . handler . timeout_exception <EOL> close_connection = False <EOL> self . _socket = None <EOL> if self . _rw_server : <EOL> self . logger . log ( BLATHER , <EOL> "<STR_LIT>" , host , port ) <EOL> host , port = self . _rw_server <EOL> self . _rw_server = None <EOL> if client . _state != KeeperState . CONNECTING : <EOL> client . _session_callback ( KeeperState . CONNECTING ) <EOL> try : <EOL> read_timeout , connect_timeout = self . _connect ( host , port ) <EOL> read_timeout = read_timeout / <NUM_LIT> <EOL> connect_timeout = connect_timeout / <NUM_LIT> <EOL> retry . reset ( ) <EOL> self . _xid = <NUM_LIT:0> <EOL> while not close_connection : <EOL> jitter_time = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) / <NUM_LIT> <EOL> timeout = max ( [ read_timeout / <NUM_LIT> - jitter_time , jitter_time ] ) <EOL> s = self . handler . select ( [ self . _socket , self . _read_pipe ] , <EOL> [ ] , [ ] , timeout ) [ <NUM_LIT:0> ] <EOL> if not s : <EOL> if self . ping_outstanding . is_set ( ) : <EOL> self . ping_outstanding . clear ( ) <EOL> raise ConnectionDropped ( <EOL> "<STR_LIT>" ) <EOL> self . _send_ping ( connect_timeout ) <EOL> elif s [ <NUM_LIT:0> ] == self . _socket : <EOL> response = self . _read_socket ( read_timeout ) <EOL> close_connection = response == CLOSE_RESPONSE <EOL> else : <EOL> self . _send_request ( read_timeout , connect_timeout ) <EOL> self . logger . info ( '<STR_LIT>' , host , port ) <EOL> client . _session_callback ( KeeperState . CLOSED ) <EOL> return STOP_CONNECTING <EOL> except ( ConnectionDropped , TimeoutError ) as e : <EOL> if isinstance ( e , ConnectionDropped ) : <EOL> self . logger . warning ( '<STR_LIT>' , e ) <EOL> else : <EOL> self . logger . warning ( '<STR_LIT>' ) <EOL> if client . _state != KeeperState . CONNECTING : <EOL> self . logger . warning ( "<STR_LIT>" ) <EOL> client . _session_callback ( KeeperState . CONNECTING ) <EOL> except AuthFailedError : <EOL> retry . reset ( ) <EOL> self . logger . warning ( '<STR_LIT>' ) <EOL> client . _session_callback ( KeeperState . AUTH_FAILED ) <EOL> return STOP_CONNECTING <EOL> except SessionExpiredError : <EOL> retry . reset ( ) <EOL> self . logger . warning ( '<STR_LIT>' ) <EOL> client . _session_callback ( KeeperState . EXPIRED_SESSION ) <EOL> except RWServerAvailable : <EOL> retry . reset ( ) <EOL> self . logger . warning ( '<STR_LIT>' ) <EOL> client . _session_callback ( KeeperState . CONNECTING ) <EOL> except Exception : <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> raise <EOL> finally : <EOL> if self . _socket is not None : <EOL> self . _socket . close ( ) <EOL> def _connect ( self , host , port ) : <EOL> client = self . client <EOL> self . logger . info ( '<STR_LIT>' , host , port ) <EOL> self . logger . log ( BLATHER , <EOL> '<STR_LIT>' , <EOL> client . _session_id , <EOL> hexlify ( client . _session_passwd ) ) <EOL> with self . _socket_error_handling ( ) : <EOL> self . _socket = self . handler . create_connection ( <EOL> ( host , port ) , client . _session_timeout / <NUM_LIT> ) <EOL> self . _socket . setblocking ( <NUM_LIT:0> ) <EOL> connect = Connect ( <NUM_LIT:0> , client . last_zxid , client . _session_timeout , <EOL> client . _session_id or <NUM_LIT:0> , client . _session_passwd , <EOL> client . read_only ) <EOL> connect_result , zxid = self . _invoke ( client . _session_timeout , connect ) <EOL> if connect_result . time_out <= <NUM_LIT:0> : <EOL> raise SessionExpiredError ( "<STR_LIT>" ) <EOL> if zxid : <EOL> client . last_zxid = zxid <EOL> client . _session_id = connect_result . session_id <EOL> client . _protocol_version = connect_result . protocol_version <EOL> negotiated_session_timeout = connect_result . time_out <EOL> connect_timeout = negotiated_session_timeout / len ( client . hosts ) <EOL> read_timeout = negotiated_session_timeout * <NUM_LIT> / <NUM_LIT> <EOL> client . _session_passwd = connect_result . passwd <EOL> self . logger . log ( BLATHER , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , client . _session_id , <EOL> hexlify ( client . _session_passwd ) , <EOL> negotiated_session_timeout , connect_timeout , <EOL> read_timeout ) <EOL> if self . sasl_server_principal : <EOL> self . _authenticate_with_sasl ( host , connect_timeout / <NUM_LIT> ) <EOL> if connect_result . read_only : <EOL> client . _session_callback ( KeeperState . CONNECTED_RO ) <EOL> self . _ro_mode = iter ( self . _server_pinger ( ) ) <EOL> else : <EOL> client . _session_callback ( KeeperState . CONNECTED ) <EOL> self . _ro_mode = None <EOL> for scheme , auth in client . auth_data : <EOL> ap = Auth ( <NUM_LIT:0> , scheme , auth ) <EOL> zxid = self . _invoke ( connect_timeout , ap , xid = AUTH_XID ) <EOL> if zxid : <EOL> client . last_zxid = zxid <EOL> return read_timeout , connect_timeout <EOL> def _authenticate_with_sasl ( self , host , timeout ) : <EOL> saslc = sasl . Client ( ) <EOL> saslc . setAttr ( '<STR_LIT:host>' , str ( host ) ) <EOL> saslc . setAttr ( '<STR_LIT>' , str ( self . sasl_server_principal ) ) <EOL> saslc . init ( ) <EOL> ret , chosen_mech , initial_response = saslc . start ( '<STR_LIT>' ) <EOL> if not ret : <EOL> raise SaslException ( saslc . getError ( ) ) <EOL> response = initial_response <EOL> xid = <NUM_LIT:0> <EOL> while True : <EOL> xid += <NUM_LIT:1> <EOL> request = SASL ( response ) <EOL> self . _submit ( request , timeout , xid ) <EOL> header , buffer , offset = self . _read_header ( timeout ) <EOL> if header . xid != xid : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' , xid , header . xid ) <EOL> if header . zxid > <NUM_LIT:0> : <EOL> client . last_zxid = zxid <EOL> if header . err : <EOL> callback_exception = EXCEPTIONS [ header . err ] ( ) <EOL> self . logger . debug ( <EOL> '<STR_LIT>' , xid , callback_exception ) <EOL> raise callback_exception <EOL> token , _ = SASL . deserialize ( buffer , offset ) <EOL> if not token : <EOL> break <EOL> ret , response = saslc . step ( token ) <EOL> if not ret : <EOL> raise SaslException ( saslc . getError ( ) ) </s>
