<s> from pypy . interpreter . baseobjspace import Wrappable <EOL> from pypy . interpreter . typedef import TypeDef , make_weakref_descr , interp_attrproperty <EOL> from pypy . interpreter . gateway import ObjSpace , W_Root , NoneNotWrapped <EOL> from pypy . interpreter . gateway import interp2app <EOL> from pypy . rlib . rarithmetic import intmask <EOL> from pypy . rlib . rsocket import RSocket , AF_INET , SOCK_STREAM <EOL> from pypy . rlib . rsocket import SocketError , SocketErrorWithErrno <EOL> from pypy . interpreter . error import OperationError <EOL> from pypy . interpreter import gateway <EOL> class W_RSocket ( Wrappable , RSocket ) : <EOL> def __del__ ( self ) : <EOL> self . clear_all_weakrefs ( ) <EOL> self . close ( ) <EOL> def accept_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> sock , addr = self . accept ( W_RSocket ) <EOL> return space . newtuple ( [ space . wrap ( sock ) , <EOL> addr . as_object ( space ) ] ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> accept_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def bind_w ( self , space , w_addr ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . bind ( self . addr_from_object ( space , w_addr ) ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> bind_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root ] <EOL> def close_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . close ( ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> close_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def connect_w ( self , space , w_addr ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . connect ( self . addr_from_object ( space , w_addr ) ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> except TypeError , e : <EOL> raise OperationError ( space . w_TypeError , <EOL> space . wrap ( str ( e ) ) ) <EOL> connect_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root ] <EOL> def connect_ex_w ( self , space , w_addr ) : <EOL> """<STR_LIT>""" <EOL> error = self . connect_ex ( self . addr_from_object ( space , w_addr ) ) <EOL> return space . wrap ( error ) <EOL> connect_ex_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root ] <EOL> def dup_w ( self , space ) : <EOL> try : <EOL> return self . dup ( W_RSocket ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> dup_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def fileno_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> fd = self . fileno ( ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return space . wrap ( intmask ( fd ) ) <EOL> fileno_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def getpeername_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> addr = self . getpeername ( ) <EOL> return addr . as_object ( space ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> getpeername_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def getsockname_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> addr = self . getsockname ( ) <EOL> return addr . as_object ( space ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> getsockname_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def getsockopt_w ( self , space , level , optname , w_buflen = NoneNotWrapped ) : <EOL> """<STR_LIT>""" <EOL> if w_buflen is None : <EOL> try : <EOL> return space . wrap ( self . getsockopt_int ( level , optname ) ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> buflen = space . int_w ( w_buflen ) <EOL> return space . wrap ( self . getsockopt ( level , optname , buflen ) ) <EOL> getsockopt_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , int , int , W_Root ] <EOL> def gettimeout_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> timeout = self . gettimeout ( ) <EOL> if timeout < <NUM_LIT:0.0> : <EOL> return space . w_None <EOL> return space . wrap ( timeout ) <EOL> gettimeout_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> def listen_w ( self , space , backlog ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . listen ( backlog ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> listen_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , int ] <EOL> def makefile_w ( self , space , w_mode = "<STR_LIT:r>" , w_buffsize = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> return app_makefile ( space , self , w_mode , w_buffsize ) <EOL> makefile_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root , W_Root ] <EOL> def recv_w ( self , space , buffersize , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> data = self . recv ( buffersize , flags ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return space . wrap ( data ) <EOL> recv_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , '<STR_LIT>' , int ] <EOL> def recvfrom_w ( self , space , buffersize , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> data , addr = self . recvfrom ( buffersize , flags ) <EOL> if addr : <EOL> w_addr = addr . as_object ( space ) <EOL> else : <EOL> w_addr = space . w_None <EOL> return space . newtuple ( [ space . wrap ( data ) , w_addr ] ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> recvfrom_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , '<STR_LIT>' , int ] <EOL> def send_w ( self , space , data , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> count = self . send ( data , flags ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return space . wrap ( count ) <EOL> send_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , '<STR_LIT>' , int ] <EOL> def sendall_w ( self , space , data , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> count = self . sendall ( data , flags ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> sendall_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , '<STR_LIT>' , int ] <EOL> def sendto_w ( self , space , data , w_param2 , w_param3 = NoneNotWrapped ) : <EOL> """<STR_LIT>""" <EOL> if w_param3 is None : <EOL> flags = <NUM_LIT:0> <EOL> w_addr = w_param2 <EOL> else : <EOL> flags = space . int_w ( w_param2 ) <EOL> w_addr = w_param3 <EOL> try : <EOL> addr = self . addr_from_object ( space , w_addr ) <EOL> count = self . sendto ( data , flags , addr ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return space . wrap ( count ) <EOL> sendto_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , '<STR_LIT>' , W_Root , W_Root ] <EOL> def setblocking_w ( self , space , flag ) : <EOL> """<STR_LIT>""" <EOL> self . setblocking ( bool ( flag ) ) <EOL> setblocking_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , int ] <EOL> def setsockopt_w ( self , space , level , optname , w_optval ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> optval = space . int_w ( w_optval ) <EOL> except : <EOL> optval = space . str_w ( w_optval ) <EOL> try : <EOL> self . setsockopt ( level , optname , optval ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return <EOL> try : <EOL> self . setsockopt_int ( level , optname , optval ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> setsockopt_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , int , int , W_Root ] <EOL> def settimeout_w ( self , space , w_timeout ) : <EOL> """<STR_LIT>""" <EOL> if space . is_w ( w_timeout , space . w_None ) : <EOL> timeout = - <NUM_LIT:1.0> <EOL> else : <EOL> timeout = space . float_w ( w_timeout ) <EOL> if timeout < <NUM_LIT:0.0> : <EOL> raise OperationError ( space . w_ValueError , <EOL> space . wrap ( '<STR_LIT>' ) ) <EOL> self . settimeout ( timeout ) <EOL> settimeout_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root ] <EOL> def recv_into_w ( self , space , w_buffer , nbytes = <NUM_LIT:0> , flags = <NUM_LIT:0> ) : <EOL> rwbuffer = space . rwbuffer_w ( w_buffer ) <EOL> lgt = rwbuffer . getlength ( ) <EOL> if nbytes == <NUM_LIT:0> or nbytes > lgt : <EOL> nbytes = lgt <EOL> try : <EOL> return space . wrap ( self . recvinto ( rwbuffer , nbytes , flags ) ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> recv_into_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root , int , int ] <EOL> def recvfrom_into_w ( self , space , w_buffer , nbytes = <NUM_LIT:0> , flags = <NUM_LIT:0> ) : <EOL> rwbuffer = space . rwbuffer_w ( w_buffer ) <EOL> lgt = rwbuffer . getlength ( ) <EOL> if nbytes == <NUM_LIT:0> or nbytes > lgt : <EOL> nbytes = lgt <EOL> try : <EOL> readlgt , addr = self . recvfrom_into ( rwbuffer , nbytes , flags ) <EOL> if addr : <EOL> w_addr = addr . as_object ( space ) <EOL> else : <EOL> w_addr = space . w_None <EOL> return space . newtuple ( [ space . wrap ( readlgt ) , w_addr ] ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> recvfrom_into_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , W_Root , int , int ] <EOL> def shutdown_w ( self , space , how ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . shutdown ( how ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> shutdown_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace , int ] <EOL> usecount = <NUM_LIT:1> <EOL> def _reuse_w ( self ) : <EOL> """<STR_LIT>""" <EOL> self . usecount += <NUM_LIT:1> <EOL> _reuse_w . unwrap_spec = [ '<STR_LIT>' ] <EOL> def _drop_w ( self , space ) : <EOL> """<STR_LIT>""" <EOL> self . usecount -= <NUM_LIT:1> <EOL> if self . usecount > <NUM_LIT:0> : <EOL> return <EOL> self . close_w ( space ) <EOL> _drop_w . unwrap_spec = [ '<STR_LIT>' , ObjSpace ] <EOL> app_makefile = gateway . applevel ( r'''<STR_LIT>''' , filename = __file__ ) . interphook ( '<STR_LIT>' ) <EOL> def newsocket ( space , w_subtype , family = AF_INET , <EOL> type = SOCK_STREAM , proto = <NUM_LIT:0> ) : <EOL> try : <EOL> sock = W_RSocket ( family , type , proto ) <EOL> except SocketError , e : <EOL> raise converted_error ( space , e ) <EOL> return space . wrap ( sock ) <EOL> descr_socket_new = interp2app ( newsocket , <EOL> unwrap_spec = [ ObjSpace , W_Root , int , int , int ] ) <EOL> def converted_error ( space , e ) : <EOL> message = e . get_msg ( ) <EOL> w_module = space . getbuiltinmodule ( '<STR_LIT>' ) <EOL> w_exception_class = space . getattr ( w_module , space . wrap ( e . applevelerrcls ) ) <EOL> if isinstance ( e , SocketErrorWithErrno ) : <EOL> w_exception = space . call_function ( w_exception_class , space . wrap ( e . errno ) , <EOL> space . wrap ( message ) ) <EOL> else : <EOL> w_exception = space . call_function ( w_exception_class , space . wrap ( message ) ) <EOL> return OperationError ( w_exception_class , w_exception ) <EOL> socketmethodnames = """<STR_LIT>""" . split ( ) <EOL> for name in ( '<STR_LIT>' , ) : <EOL> if not hasattr ( RSocket , name ) : <EOL> socketmethodnames . remove ( name ) <EOL> socketmethods = { } <EOL> for methodname in socketmethodnames : <EOL> method = getattr ( W_RSocket , methodname + '<STR_LIT>' ) <EOL> assert hasattr ( method , '<STR_LIT>' ) , methodname <EOL> assert method . im_func . func_code . co_argcount == len ( method . unwrap_spec ) , methodname <EOL> socketmethods [ methodname ] = interp2app ( method , unwrap_spec = method . unwrap_spec ) <EOL> W_RSocket . typedef = TypeDef ( "<STR_LIT>" , <EOL> __doc__ = """<STR_LIT>""" , <EOL> __new__ = descr_socket_new , <EOL> __weakref__ = make_weakref_descr ( W_RSocket ) , <EOL> type = interp_attrproperty ( '<STR_LIT:type>' , W_RSocket ) , <EOL> proto = interp_attrproperty ( '<STR_LIT>' , W_RSocket ) , <EOL> family = interp_attrproperty ( '<STR_LIT>' , W_RSocket ) , <EOL> ** socketmethods <EOL> ) </s>
