<s> """<STR_LIT>""" <EOL> import os <EOL> import select <EOL> from time import time <EOL> from collections import defaultdict , deque <EOL> from errno import EAGAIN , EALREADY , EBADF <EOL> from errno import ECONNABORTED , EINPROGRESS , EINTR , EISCONN , EMFILE , ENFILE <EOL> from errno import ENOBUFS , ENOMEM , ENOTCONN , EPERM , EPIPE , EINVAL , EWOULDBLOCK <EOL> from _socket import socket as SocketType <EOL> from socket import gaierror <EOL> from socket import error as SocketError <EOL> from socket import getfqdn , gethostbyname , socket , getaddrinfo , gethostname <EOL> from socket import AF_INET , AF_INET6 , IPPROTO_TCP , SOCK_STREAM , SOCK_DGRAM <EOL> from socket import SOL_SOCKET , SO_BROADCAST , SO_REUSEADDR , TCP_NODELAY <EOL> try : <EOL> from ssl import wrap_socket as ssl_socket <EOL> from ssl import CERT_NONE , PROTOCOL_SSLv23 <EOL> from ssl import SSLError , SSL_ERROR_WANT_WRITE , SSL_ERROR_WANT_READ <EOL> HAS_SSL = <NUM_LIT:1> <EOL> except ImportError : <EOL> import warnings <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> HAS_SSL = <NUM_LIT:0> <EOL> from circuits . six import binary_type <EOL> from circuits . core . utils import findcmp <EOL> from circuits . core import handler , BaseComponent <EOL> from circuits . core . pollers import BasePoller , Poller <EOL> from . events import close , closed , connect , connected , disconnect , disconnected , error , read , ready , write , unreachable <EOL> BUFSIZE = <NUM_LIT> <EOL> BACKLOG = <NUM_LIT> <EOL> def do_handshake ( sock , on_done = None , on_error = None , extra_args = None ) : <EOL> """<STR_LIT>""" <EOL> extra_args = extra_args or ( ) <EOL> while True : <EOL> try : <EOL> sock . do_handshake ( ) <EOL> break <EOL> except SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> select . select ( [ sock ] , [ ] , [ ] ) <EOL> elif err . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> select . select ( [ ] , [ sock ] , [ ] ) <EOL> else : <EOL> callable ( on_error ) and on_error ( sock , err ) <EOL> return <EOL> yield <EOL> callable ( on_done ) and on_done ( sock , * extra_args ) <EOL> class Client ( BaseComponent ) : <EOL> channel = "<STR_LIT>" <EOL> def __init__ ( self , bind = None , bufsize = BUFSIZE , channel = channel , ** kwargs ) : <EOL> super ( Client , self ) . __init__ ( channel = channel , ** kwargs ) <EOL> if isinstance ( bind , SocketType ) : <EOL> self . _bind = bind . getsockname ( ) <EOL> self . _sock = bind <EOL> else : <EOL> self . _bind = self . parse_bind_parameter ( bind ) <EOL> self . _sock = self . _create_socket ( ) <EOL> self . _bufsize = bufsize <EOL> self . _ssock = None <EOL> self . _poller = None <EOL> self . _buffer = deque ( ) <EOL> self . _closeflag = False <EOL> self . _connected = False <EOL> self . host = None <EOL> self . port = <NUM_LIT:0> <EOL> self . secure = False <EOL> self . server = { } <EOL> self . issuer = { } <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv4_parameter ( bind_parameter ) <EOL> @ property <EOL> def connected ( self ) : <EOL> return getattr ( self , "<STR_LIT>" , None ) <EOL> @ handler ( "<STR_LIT>" , "<STR_LIT>" , channel = "<STR_LIT:*>" ) <EOL> def _on_registered_or_started ( self , component , manager = None ) : <EOL> if self . _poller is None : <EOL> if isinstance ( component , BasePoller ) : <EOL> self . _poller = component <EOL> self . fire ( ready ( self ) ) <EOL> else : <EOL> if component is not self : <EOL> return <EOL> component = findcmp ( self . root , BasePoller ) <EOL> if component is not None : <EOL> self . _poller = component <EOL> self . fire ( ready ( self ) ) <EOL> else : <EOL> self . _poller = Poller ( ) . register ( self ) <EOL> self . fire ( ready ( self ) ) <EOL> @ handler ( "<STR_LIT>" , channel = "<STR_LIT:*>" ) <EOL> def _on_stopped ( self , component ) : <EOL> self . fire ( close ( ) ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def _on_read_value_changed ( self , value ) : <EOL> if isinstance ( value , binary_type ) : <EOL> self . fire ( write ( value ) ) <EOL> @ handler ( "<STR_LIT>" , channel = "<STR_LIT:*>" ) <EOL> def _on_prepare_unregister ( self , event , c ) : <EOL> if event . in_subtree ( self ) : <EOL> self . _close ( ) <EOL> def _close ( self ) : <EOL> if not self . _connected : <EOL> return <EOL> self . _poller . discard ( self . _sock ) <EOL> self . _buffer . clear ( ) <EOL> self . _closeflag = False <EOL> self . _connected = False <EOL> try : <EOL> self . _sock . shutdown ( <NUM_LIT:2> ) <EOL> self . _sock . close ( ) <EOL> except SocketError : <EOL> pass <EOL> self . fire ( disconnected ( ) ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def close ( self ) : <EOL> if not self . _buffer : <EOL> self . _close ( ) <EOL> elif not self . _closeflag : <EOL> self . _closeflag = True <EOL> def _read ( self ) : <EOL> try : <EOL> if self . secure and self . _ssock : <EOL> data = self . _ssock . read ( self . _bufsize ) <EOL> else : <EOL> try : <EOL> data = self . _sock . recv ( self . _bufsize ) <EOL> except SSLError as exc : <EOL> if exc . errno in ( SSL_ERROR_WANT_READ , SSL_ERROR_WANT_WRITE ) : <EOL> return <EOL> raise <EOL> if data : <EOL> self . fire ( read ( data ) ) . notify = True <EOL> else : <EOL> self . close ( ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] == EWOULDBLOCK : <EOL> return <EOL> else : <EOL> self . fire ( error ( e ) ) <EOL> self . _close ( ) <EOL> def _write ( self , data ) : <EOL> try : <EOL> if self . secure and self . _ssock : <EOL> nbytes = self . _ssock . write ( data ) <EOL> else : <EOL> nbytes = self . _sock . send ( data ) <EOL> if nbytes < len ( data ) : <EOL> self . _buffer . appendleft ( data [ nbytes : ] ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] in ( EPIPE , ENOTCONN ) : <EOL> self . _close ( ) <EOL> else : <EOL> self . fire ( error ( e ) ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def write ( self , data ) : <EOL> if not self . _poller . isWriting ( self . _sock ) : <EOL> self . _poller . addWriter ( self , self . _sock ) <EOL> self . _buffer . append ( data ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def __on_disconnect ( self , sock ) : <EOL> self . _close ( ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def __on_read ( self , sock ) : <EOL> self . _read ( ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def __on_write ( self , sock ) : <EOL> if self . _buffer : <EOL> data = self . _buffer . popleft ( ) <EOL> self . _write ( data ) <EOL> if not self . _buffer : <EOL> if self . _closeflag : <EOL> self . _close ( ) <EOL> elif self . _poller . isWriting ( self . _sock ) : <EOL> self . _poller . removeWriter ( self . _sock ) <EOL> class TCPClient ( Client ) : <EOL> socket_family = AF_INET <EOL> def init ( self , connect_timeout = <NUM_LIT:5> , * args , ** kwargs ) : <EOL> self . connect_timeout = connect_timeout <EOL> def _create_socket ( self ) : <EOL> sock = socket ( self . socket_family , SOCK_STREAM , IPPROTO_TCP ) <EOL> if self . _bind is not None : <EOL> sock . bind ( self . _bind ) <EOL> sock . setblocking ( False ) <EOL> sock . setsockopt ( IPPROTO_TCP , TCP_NODELAY , <NUM_LIT:1> ) <EOL> return sock <EOL> @ handler ( "<STR_LIT>" ) <EOL> def connect ( self , host , port , secure = False , ** kwargs ) : <EOL> self . host = host <EOL> self . port = port <EOL> self . secure = secure <EOL> if self . secure : <EOL> self . certfile = kwargs . get ( "<STR_LIT>" , None ) <EOL> self . keyfile = kwargs . get ( "<STR_LIT>" , None ) <EOL> self . ca_certs = kwargs . get ( "<STR_LIT>" , None ) <EOL> try : <EOL> r = self . _sock . connect ( ( host , port ) ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] in ( EBADF , EINVAL , ) : <EOL> self . _sock = self . _create_socket ( ) <EOL> r = self . _sock . connect_ex ( ( host , port ) ) <EOL> else : <EOL> r = e . args [ <NUM_LIT:0> ] <EOL> if r not in ( EISCONN , EWOULDBLOCK , EINPROGRESS , EALREADY ) : <EOL> self . fire ( unreachable ( host , port , e ) ) <EOL> self . fire ( error ( e ) ) <EOL> self . _close ( ) <EOL> raise StopIteration ( ) <EOL> stop_time = time ( ) + self . connect_timeout <EOL> while time ( ) < stop_time : <EOL> try : <EOL> self . _sock . getpeername ( ) <EOL> self . _connected = True <EOL> break <EOL> except Exception as e : <EOL> yield <EOL> if not self . _connected : <EOL> self . fire ( unreachable ( host , port ) ) <EOL> raise StopIteration ( ) <EOL> def on_done ( sock ) : <EOL> self . _poller . addReader ( self , sock ) <EOL> self . fire ( connected ( host , port ) ) <EOL> if self . secure : <EOL> def on_error ( sock , err ) : <EOL> self . fire ( error ( sock , err ) ) <EOL> self . _close ( ) <EOL> self . _sock = ssl_socket ( <EOL> self . _sock , self . keyfile , self . certfile , ca_certs = self . ca_certs , <EOL> do_handshake_on_connect = False <EOL> ) <EOL> for _ in do_handshake ( self . _sock , on_done , on_error ) : <EOL> yield <EOL> else : <EOL> on_done ( self . _sock ) <EOL> class TCP6Client ( TCPClient ) : <EOL> socket_family = AF_INET6 <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv6_parameter ( bind_parameter ) <EOL> class UNIXClient ( Client ) : <EOL> def _create_socket ( self ) : <EOL> from socket import AF_UNIX <EOL> sock = socket ( AF_UNIX , SOCK_STREAM ) <EOL> if self . _bind is not None : <EOL> sock . bind ( self . _bind ) <EOL> sock . setblocking ( False ) <EOL> return sock <EOL> @ handler ( "<STR_LIT>" ) <EOL> def ready ( self , component ) : <EOL> if self . _poller is not None and self . _connected : <EOL> self . _poller . addReader ( self , self . _sock ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def connect ( self , path , secure = False , ** kwargs ) : <EOL> self . path = path <EOL> self . secure = secure <EOL> if self . secure : <EOL> self . certfile = kwargs . get ( "<STR_LIT>" , None ) <EOL> self . keyfile = kwargs . get ( "<STR_LIT>" , None ) <EOL> self . ca_certs = kwargs . get ( "<STR_LIT>" , None ) <EOL> try : <EOL> r = self . _sock . connect_ex ( path ) <EOL> except SocketError as e : <EOL> r = e . args [ <NUM_LIT:0> ] <EOL> if r : <EOL> if r in ( EISCONN , EWOULDBLOCK , EINPROGRESS , EALREADY ) : <EOL> self . _connected = True <EOL> else : <EOL> self . fire ( error ( r ) ) <EOL> return <EOL> self . _connected = True <EOL> self . _poller . addReader ( self , self . _sock ) <EOL> if self . secure : <EOL> def on_done ( sock ) : <EOL> self . fire ( connected ( gethostname ( ) , path ) ) <EOL> def on_error ( sock , err ) : <EOL> self . fire ( error ( err ) ) <EOL> self . _ssock = ssl_socket ( <EOL> self . _sock , self . keyfile , self . certfile , ca_certs = self . ca_certs , <EOL> do_handshake_on_connect = False <EOL> ) <EOL> for _ in do_handshake ( self . _ssock , on_done , on_error ) : <EOL> yield <EOL> else : <EOL> self . fire ( connected ( gethostname ( ) , path ) ) <EOL> class Server ( BaseComponent ) : <EOL> channel = "<STR_LIT>" <EOL> def __init__ ( self , bind , secure = False , backlog = BACKLOG , <EOL> bufsize = BUFSIZE , channel = channel , ** kwargs ) : <EOL> super ( Server , self ) . __init__ ( channel = channel ) <EOL> self . _bind = self . parse_bind_parameter ( bind ) <EOL> self . _backlog = backlog <EOL> self . _bufsize = bufsize <EOL> if isinstance ( bind , socket ) : <EOL> self . _sock = bind <EOL> else : <EOL> self . _sock = self . _create_socket ( ) <EOL> self . _closeq = [ ] <EOL> self . _clients = [ ] <EOL> self . _poller = None <EOL> self . _buffers = defaultdict ( deque ) <EOL> self . secure = secure <EOL> if self . secure : <EOL> try : <EOL> self . certfile = kwargs [ "<STR_LIT>" ] <EOL> except KeyError : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" ) <EOL> self . keyfile = kwargs . get ( "<STR_LIT>" , None ) <EOL> self . cert_reqs = kwargs . get ( "<STR_LIT>" , CERT_NONE ) <EOL> self . ssl_version = kwargs . get ( "<STR_LIT>" , PROTOCOL_SSLv23 ) <EOL> self . ca_certs = kwargs . get ( "<STR_LIT>" , None ) <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv4_parameter ( bind_parameter ) <EOL> @ property <EOL> def connected ( self ) : <EOL> return True <EOL> @ property <EOL> def host ( self ) : <EOL> if getattr ( self , "<STR_LIT>" , None ) is not None : <EOL> try : <EOL> sockname = self . _sock . getsockname ( ) <EOL> if isinstance ( sockname , tuple ) : <EOL> return sockname [ <NUM_LIT:0> ] <EOL> else : <EOL> return sockname <EOL> except SocketError : <EOL> return None <EOL> @ property <EOL> def port ( self ) : <EOL> if getattr ( self , "<STR_LIT>" , None ) is not None : <EOL> try : <EOL> sockname = self . _sock . getsockname ( ) <EOL> if isinstance ( sockname , tuple ) : <EOL> return sockname [ <NUM_LIT:1> ] <EOL> except SocketError : <EOL> return None <EOL> @ handler ( "<STR_LIT>" , "<STR_LIT>" , channel = "<STR_LIT:*>" ) <EOL> def _on_registered_or_started ( self , component , manager = None ) : <EOL> if self . _poller is None : <EOL> if isinstance ( component , BasePoller ) : <EOL> self . _poller = component <EOL> self . _poller . addReader ( self , self . _sock ) <EOL> self . fire ( ready ( self , ( self . host , self . port ) ) ) <EOL> else : <EOL> if component is not self : <EOL> return <EOL> component = findcmp ( self . root , BasePoller ) <EOL> if component is not None : <EOL> self . _poller = component <EOL> self . _poller . addReader ( self , self . _sock ) <EOL> self . fire ( ready ( self , ( self . host , self . port ) ) ) <EOL> else : <EOL> self . _poller = Poller ( ) . register ( self ) <EOL> self . _poller . addReader ( self , self . _sock ) <EOL> self . fire ( ready ( self , ( self . host , self . port ) ) ) <EOL> @ handler ( "<STR_LIT>" , channel = "<STR_LIT:*>" ) <EOL> def _on_stopped ( self , component ) : <EOL> self . fire ( close ( ) ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def _on_read_value_changed ( self , value ) : <EOL> if isinstance ( value . value , binary_type ) : <EOL> sock = value . event . args [ <NUM_LIT:0> ] <EOL> self . fire ( write ( sock , value . value ) ) <EOL> def _close ( self , sock ) : <EOL> if sock is None : <EOL> return <EOL> if sock != self . _sock and sock not in self . _clients : <EOL> return <EOL> self . _poller . discard ( sock ) <EOL> if sock in self . _buffers : <EOL> del self . _buffers [ sock ] <EOL> if sock in self . _clients : <EOL> self . _clients . remove ( sock ) <EOL> else : <EOL> self . _sock = None <EOL> try : <EOL> sock . shutdown ( <NUM_LIT:2> ) <EOL> sock . close ( ) <EOL> except SocketError : <EOL> pass <EOL> self . fire ( disconnect ( sock ) ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def close ( self , sock = None ) : <EOL> is_closed = sock is None <EOL> if sock is None : <EOL> socks = [ self . _sock ] <EOL> socks . extend ( self . _clients [ : ] ) <EOL> else : <EOL> socks = [ sock ] <EOL> for sock in socks : <EOL> if not self . _buffers [ sock ] : <EOL> self . _close ( sock ) <EOL> elif sock not in self . _closeq : <EOL> self . _closeq . append ( sock ) <EOL> if is_closed : <EOL> self . fire ( closed ( ) ) <EOL> def _read ( self , sock ) : <EOL> if sock not in self . _clients : <EOL> return <EOL> try : <EOL> data = sock . recv ( self . _bufsize ) <EOL> if data : <EOL> self . fire ( read ( sock , data ) ) . notify = True <EOL> else : <EOL> self . close ( sock ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] == EWOULDBLOCK : <EOL> return <EOL> else : <EOL> self . fire ( error ( sock , e ) ) <EOL> self . _close ( sock ) <EOL> def _write ( self , sock , data ) : <EOL> if sock not in self . _clients : <EOL> return <EOL> try : <EOL> nbytes = sock . send ( data ) <EOL> if nbytes < len ( data ) : <EOL> self . _buffers [ sock ] . appendleft ( data [ nbytes : ] ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] not in ( EINTR , EWOULDBLOCK , ENOBUFS ) : <EOL> self . fire ( error ( sock , e ) ) <EOL> self . _close ( sock ) <EOL> else : <EOL> self . _buffers [ sock ] . appendleft ( data ) <EOL> @ handler ( "<STR_LIT>" ) <EOL> def write ( self , sock , data ) : <EOL> if not self . _poller . isWriting ( sock ) : <EOL> self . _poller . addWriter ( self , sock ) <EOL> self . _buffers [ sock ] . append ( data ) <EOL> def _accept ( self ) : <EOL> def on_done ( sock , host ) : <EOL> sock . setblocking ( False ) <EOL> self . _poller . addReader ( self , sock ) <EOL> self . _clients . append ( sock ) <EOL> self . fire ( connect ( sock , * host ) ) <EOL> def on_error ( sock , err ) : <EOL> self . fire ( error ( sock , err ) ) <EOL> self . _close ( sock ) <EOL> try : <EOL> newsock , host = self . _sock . accept ( ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] in ( EWOULDBLOCK , EAGAIN ) : <EOL> return <EOL> elif e . args [ <NUM_LIT:0> ] == EPERM : <EOL> return <EOL> elif e . args [ <NUM_LIT:0> ] in ( EMFILE , ENOBUFS , ENFILE , ENOMEM , ECONNABORTED ) : <EOL> return <EOL> else : <EOL> raise <EOL> if self . secure and HAS_SSL : <EOL> sslsock = ssl_socket ( <EOL> newsock , <EOL> server_side = True , <EOL> keyfile = self . keyfile , <EOL> ca_certs = self . ca_certs , <EOL> certfile = self . certfile , <EOL> cert_reqs = self . cert_reqs , <EOL> ssl_version = self . ssl_version , <EOL> do_handshake_on_connect = False <EOL> ) <EOL> for _ in do_handshake ( sslsock , on_done , on_error , extra_args = ( host , ) ) : <EOL> yield <EOL> else : <EOL> on_done ( newsock , host ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def _on_disconnect ( self , sock ) : <EOL> self . _close ( sock ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def _on_read ( self , sock ) : <EOL> if sock == self . _sock : <EOL> return self . _accept ( ) <EOL> else : <EOL> self . _read ( sock ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> ) <EOL> def _on_write ( self , sock ) : <EOL> if self . _buffers [ sock ] : <EOL> data = self . _buffers [ sock ] . popleft ( ) <EOL> self . _write ( sock , data ) <EOL> if not self . _buffers [ sock ] : <EOL> if sock in self . _closeq : <EOL> self . _closeq . remove ( sock ) <EOL> self . _close ( sock ) <EOL> elif self . _poller . isWriting ( sock ) : <EOL> self . _poller . removeWriter ( sock ) <EOL> class TCPServer ( Server ) : <EOL> socket_family = AF_INET <EOL> def _create_socket ( self ) : <EOL> sock = socket ( self . socket_family , SOCK_STREAM ) <EOL> sock . setsockopt ( SOL_SOCKET , SO_REUSEADDR , <NUM_LIT:1> ) <EOL> sock . setsockopt ( IPPROTO_TCP , TCP_NODELAY , <NUM_LIT:1> ) <EOL> sock . setblocking ( False ) <EOL> sock . bind ( self . _bind ) <EOL> sock . listen ( self . _backlog ) <EOL> return sock <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv4_parameter ( bind_parameter ) <EOL> def parse_ipv4_parameter ( bind_parameter ) : <EOL> if isinstance ( bind_parameter , int ) : <EOL> try : <EOL> bind = ( gethostbyname ( gethostname ( ) ) , bind_parameter ) <EOL> except gaierror : <EOL> bind = ( "<STR_LIT>" , bind_parameter ) <EOL> elif isinstance ( bind_parameter , str ) and "<STR_LIT::>" in bind_parameter : <EOL> host , port = bind_parameter . split ( "<STR_LIT::>" ) <EOL> port = int ( port ) <EOL> bind = ( host , port ) <EOL> else : <EOL> bind = bind_parameter <EOL> return bind <EOL> def parse_ipv6_parameter ( bind_parameter ) : <EOL> if isinstance ( bind_parameter , int ) : <EOL> try : <EOL> _ , _ , _ , _ , bind = getaddrinfo ( getfqdn ( ) , bind_parameter , AF_INET6 ) [ <NUM_LIT:0> ] <EOL> except ( gaierror , IndexError ) : <EOL> bind = ( "<STR_LIT>" , bind_parameter ) <EOL> else : <EOL> bind = bind_parameter <EOL> return bind <EOL> class TCP6Server ( TCPServer ) : <EOL> socket_family = AF_INET6 <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv6_parameter ( bind_parameter ) <EOL> class UNIXServer ( Server ) : <EOL> def _create_socket ( self ) : <EOL> from socket import AF_UNIX <EOL> if os . path . exists ( self . _bind ) : <EOL> os . unlink ( self . _bind ) <EOL> sock = socket ( AF_UNIX , SOCK_STREAM ) <EOL> sock . bind ( self . _bind ) <EOL> sock . setsockopt ( SOL_SOCKET , SO_REUSEADDR , <NUM_LIT:1> ) <EOL> sock . setblocking ( False ) <EOL> sock . listen ( self . _backlog ) <EOL> return sock <EOL> class UDPServer ( Server ) : <EOL> socket_family = AF_INET <EOL> def _create_socket ( self ) : <EOL> sock = socket ( self . socket_family , SOCK_DGRAM ) <EOL> sock . bind ( self . _bind ) <EOL> sock . setsockopt ( SOL_SOCKET , SO_BROADCAST , <NUM_LIT:1> ) <EOL> sock . setsockopt ( SOL_SOCKET , SO_REUSEADDR , <NUM_LIT:1> ) <EOL> sock . setblocking ( False ) <EOL> return sock <EOL> def _close ( self , sock ) : <EOL> self . _poller . discard ( sock ) <EOL> if sock in self . _buffers : <EOL> del self . _buffers [ sock ] <EOL> try : <EOL> sock . shutdown ( <NUM_LIT:2> ) <EOL> except SocketError : <EOL> pass <EOL> try : <EOL> sock . close ( ) <EOL> except SocketError : <EOL> pass <EOL> self . fire ( disconnect ( sock ) ) <EOL> @ handler ( "<STR_LIT>" , override = True ) <EOL> def close ( self ) : <EOL> self . fire ( closed ( ) ) <EOL> if self . _buffers [ self . _sock ] and self . _sock not in self . _closeq : <EOL> self . _closeq . append ( self . _sock ) <EOL> else : <EOL> self . _close ( self . _sock ) <EOL> def _read ( self ) : <EOL> try : <EOL> data , address = self . _sock . recvfrom ( self . _bufsize ) <EOL> if data : <EOL> self . fire ( read ( address , data ) ) . notify = True <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] in ( EWOULDBLOCK , EAGAIN ) : <EOL> return <EOL> self . fire ( error ( self . _sock , e ) ) <EOL> self . _close ( self . _sock ) <EOL> def _write ( self , address , data ) : <EOL> try : <EOL> bytes = self . _sock . sendto ( data , address ) <EOL> if bytes < len ( data ) : <EOL> self . _buffers [ self . _sock ] . appendleft ( data [ bytes : ] ) <EOL> except SocketError as e : <EOL> if e . args [ <NUM_LIT:0> ] in ( EPIPE , ENOTCONN ) : <EOL> self . _close ( self . _sock ) <EOL> else : <EOL> self . fire ( error ( self . _sock , e ) ) <EOL> @ handler ( "<STR_LIT>" , override = True ) <EOL> def write ( self , address , data ) : <EOL> if not self . _poller . isWriting ( self . _sock ) : <EOL> self . _poller . addWriter ( self , self . _sock ) <EOL> self . _buffers [ self . _sock ] . append ( ( address , data ) ) <EOL> @ handler ( "<STR_LIT>" , override = True ) <EOL> def broadcast ( self , data , port ) : <EOL> self . write ( ( "<STR_LIT>" , port ) , data ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> , override = True ) <EOL> def _on_disconnect ( self , sock ) : <EOL> self . _close ( sock ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> , override = True ) <EOL> def _on_read ( self , sock ) : <EOL> self . _read ( ) <EOL> @ handler ( "<STR_LIT>" , priority = <NUM_LIT:1> , override = True ) <EOL> def _on_write ( self , sock ) : <EOL> if self . _buffers [ self . _sock ] : <EOL> address , data = self . _buffers [ self . _sock ] . popleft ( ) <EOL> self . _write ( address , data ) <EOL> if not self . _buffers [ self . _sock ] : <EOL> if self . _sock in self . _closeq : <EOL> self . _closeq . remove ( self . _sock ) <EOL> self . _close ( self . _sock ) <EOL> elif self . _poller . isWriting ( self . _sock ) : <EOL> self . _poller . removeWriter ( self . _sock ) <EOL> UDPClient = UDPServer <EOL> class UDP6Server ( UDPServer ) : <EOL> socket_family = AF_INET6 <EOL> def parse_bind_parameter ( self , bind_parameter ) : <EOL> return parse_ipv6_parameter ( bind_parameter ) <EOL> UDP6Client = UDP6Server <EOL> def Pipe ( * channels , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> from socket import socketpair <EOL> if not channels : <EOL> channels = ( "<STR_LIT:a>" , "<STR_LIT:b>" ) <EOL> s1 , s2 = socketpair ( ) <EOL> s1 . setblocking ( False ) <EOL> s2 . setblocking ( False ) <EOL> a = UNIXClient ( s1 , channel = channels [ <NUM_LIT:0> ] , ** kwargs ) <EOL> b = UNIXClient ( s2 , channel = channels [ <NUM_LIT:1> ] , ** kwargs ) <EOL> a . _connected = True <EOL> b . _connected = True <EOL> return a , b </s>
