abstract transport class all basetransport readtransport writetransport transport datagramtransport subprocesstransport class basetransport get optional transport information return self extra getname default def isclosingself close the transport buffered data will be flushed asynchronously no more data will be received after all buffered data is flushed the protocol s connectionlost method will eventually be called with none as its argument set a new protocol raise notimplementederror def getprotocolself interface for readonly transports slots def isreadingself pause the receiving end no data will be passed to the protocol s datareceived method until resumereading is called resume the receiving end data received will once again be passed to the protocol s datareceived method interface for writeonly transports slots def setwritebufferlimitsself highnone lownone raise notimplementederror def getwritebuffersizeself get the high and low watermarks for write flow control return a tuple low high where low and high are positive number of bytes raise notimplementederror def writeself data raise notimplementederror def writelinesself listofdata data b joinlistofdata self writedata def writeeofself raise notimplementederror def canwriteeofself close the transport immediately buffered data will be lost no more data will be received the protocol s connectionlost method will eventually be called with none as its argument interface representing a bidirectional transport there may be several implementations but typically the user does not implement new transports rather the platform provides some useful transports that are implemented using the platform s best practices the user never instantiates a transport directly they call a utility function passing it a protocol factory and other information necessary to create the transport and protocol e g eventloop createconnection or eventloop createserver the utility function will asynchronously create a transport and a protocol and hook them up by calling the protocol s connectionmade method passing it the transport the implementation here raises notimplemented for every method except writelines which calls write in a loop interface for datagram udp transports slots def sendtoself data addrnone raise notimplementederror def abortself raise notimplementederror class subprocesstransportbasetransport slots def getpidself get subprocess returncode see also http docs python org3librarysubprocesssubprocess popen returncode get transport for pipe with number fd raise notimplementederror def sendsignalself signal raise notimplementederror def terminateself raise notimplementederror def killself raise notimplementederror class flowcontrolmixintransport slots loop protocolpaused highwater lowwater def initself extranone loopnone super initextra assert loop is not none self loop loop self protocolpaused false self setwritebufferlimits def maybepauseprotocolself size self getwritebuffersize if size self highwater return if not self protocolpaused self protocolpaused true try self protocol pausewriting except systemexit keyboardinterrupt raise except baseexception as exc self loop callexceptionhandler message protocol pausewriting failed exception exc transport self protocol self protocol def mayberesumeprotocolself if self protocolpaused and self getwritebuffersize self lowwater self protocolpaused false try self protocol resumewriting except systemexit keyboardinterrupt raise except baseexception as exc self loop callexceptionhandler message protocol resumewriting failed exception exc transport self protocol self protocol def getwritebufferlimitsself return self lowwater self highwater def setwritebufferlimitsself highnone lownone if high is none if low is none high 64 1024 else high 4 low if low is none low high 4 if not high low 0 raise valueerror f high high r must be low low r must be 0 self highwater high self lowwater low def setwritebufferlimitsself highnone lownone self setwritebufferlimitshighhigh lowlow self maybepauseprotocol def getwritebuffersizeself raise notimplementederror base class for transports get optional transport information return true if the transport is closing or closed close the transport buffered data will be flushed asynchronously no more data will be received after all buffered data is flushed the protocol s connection_lost method will eventually be called with none as its argument set a new protocol return the current protocol interface for read only transports return true if the transport is receiving pause the receiving end no data will be passed to the protocol s data_received method until resume_reading is called resume the receiving end data received will once again be passed to the protocol s data_received method interface for write only transports set the high and low water limits for write flow control these two values control when to call the protocol s pause_writing and resume_writing methods if specified the low water limit must be less than or equal to the high water limit neither value can be negative the defaults are implementation specific if only the high water limit is given the low water limit defaults to an implementation specific value less than or equal to the high water limit setting high to zero forces low to zero as well and causes pause_writing to be called whenever the buffer becomes non empty setting low to zero causes resume_writing to be called only once the buffer is empty use of zero for either limit is generally sub optimal as it reduces opportunities for doing i o and computation concurrently return the current size of the write buffer get the high and low watermarks for write flow control return a tuple low high where low and high are positive number of bytes write some data bytes to the transport this does not block it buffers the data and arranges for it to be sent out asynchronously write a list or any iterable of data bytes to the transport the default implementation concatenates the arguments and calls write on the result close the write end after flushing buffered data this is like typing d into a unix program reading from stdin data may still be received return true if this transport supports write_eof false if not close the transport immediately buffered data will be lost no more data will be received the protocol s connection_lost method will eventually be called with none as its argument interface representing a bidirectional transport there may be several implementations but typically the user does not implement new transports rather the platform provides some useful transports that are implemented using the platform s best practices the user never instantiates a transport directly they call a utility function passing it a protocol factory and other information necessary to create the transport and protocol e g eventloop create_connection or eventloop create_server the utility function will asynchronously create a transport and a protocol and hook them up by calling the protocol s connection_made method passing it the transport the implementation here raises notimplemented for every method except writelines which calls write in a loop interface for datagram udp transports send data to the transport this does not block it buffers the data and arranges for it to be sent out asynchronously addr is target socket address if addr is none use target address pointed on transport creation close the transport immediately buffered data will be lost no more data will be received the protocol s connection_lost method will eventually be called with none as its argument get subprocess id get subprocess returncode see also http docs python org 3 library subprocess subprocess popen returncode get transport for pipe with number fd send signal to subprocess see also docs python org 3 library subprocess subprocess popen send_signal stop the subprocess alias for close method on posix oss the method sends sigterm to the subprocess on windows the win32 api function terminateprocess is called to stop the subprocess see also http docs python org 3 library subprocess subprocess popen terminate kill the subprocess on posix oss the function sends sigkill to the subprocess on windows kill is an alias for terminate see also http docs python org 3 library subprocess subprocess popen kill all the logic for write flow control in a mix in base class the subclass must implement get_write_buffer_size it must call _maybe_pause_protocol whenever the write buffer size increases and _maybe_resume_protocol whenever it decreases it may also override set_write_buffer_limits e g to specify different defaults the subclass constructor must call super __init__ extra this will call set_write_buffer_limits the user may call set_write_buffer_limits and get_write_buffer_size and their protocol s pause_writing and resume_writing may be called
__all__ = (
    'BaseTransport', 'ReadTransport', 'WriteTransport',
    'Transport', 'DatagramTransport', 'SubprocessTransport',
)


class BaseTransport:
    

    __slots__ = ('_extra',)

    def __init__(self, extra=None):
        if extra is None:
            extra = {}
        self._extra = extra

    def get_extra_info(self, name, default=None):
        
        return self._extra.get(name, default)

    def is_closing(self):
        
        raise NotImplementedError

    def close(self):
        
        raise NotImplementedError

    def set_protocol(self, protocol):
        
        raise NotImplementedError

    def get_protocol(self):
        
        raise NotImplementedError


class ReadTransport(BaseTransport):
    

    __slots__ = ()

    def is_reading(self):
        
        raise NotImplementedError

    def pause_reading(self):
        
        raise NotImplementedError

    def resume_reading(self):
        
        raise NotImplementedError


class WriteTransport(BaseTransport):
    

    __slots__ = ()

    def set_write_buffer_limits(self, high=None, low=None):
        
        raise NotImplementedError

    def get_write_buffer_size(self):
        
        raise NotImplementedError

    def get_write_buffer_limits(self):
        
        raise NotImplementedError

    def write(self, data):
        
        raise NotImplementedError

    def writelines(self, list_of_data):
        
        data = b''.join(list_of_data)
        self.write(data)

    def write_eof(self):
        
        raise NotImplementedError

    def can_write_eof(self):
        
        raise NotImplementedError

    def abort(self):
        
        raise NotImplementedError


class Transport(ReadTransport, WriteTransport):
    

    __slots__ = ()


class DatagramTransport(BaseTransport):
    

    __slots__ = ()

    def sendto(self, data, addr=None):
        
        raise NotImplementedError

    def abort(self):
        
        raise NotImplementedError


class SubprocessTransport(BaseTransport):

    __slots__ = ()

    def get_pid(self):
        
        raise NotImplementedError

    def get_returncode(self):
        
        raise NotImplementedError

    def get_pipe_transport(self, fd):
        
        raise NotImplementedError

    def send_signal(self, signal):
        
        raise NotImplementedError

    def terminate(self):
        
        raise NotImplementedError

    def kill(self):
        
        raise NotImplementedError


class _FlowControlMixin(Transport):
    

    __slots__ = ('_loop', '_protocol_paused', '_high_water', '_low_water')

    def __init__(self, extra=None, loop=None):
        super().__init__(extra)
        assert loop is not None
        self._loop = loop
        self._protocol_paused = False
        self._set_write_buffer_limits()

    def _maybe_pause_protocol(self):
        size = self.get_write_buffer_size()
        if size <= self._high_water:
            return
        if not self._protocol_paused:
            self._protocol_paused = True
            try:
                self._protocol.pause_writing()
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                self._loop.call_exception_handler({
                    'message': 'protocol.pause_writing() failed',
                    'exception': exc,
                    'transport': self,
                    'protocol': self._protocol,
                })

    def _maybe_resume_protocol(self):
        if (self._protocol_paused and
                self.get_write_buffer_size() <= self._low_water):
            self._protocol_paused = False
            try:
                self._protocol.resume_writing()
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                self._loop.call_exception_handler({
                    'message': 'protocol.resume_writing() failed',
                    'exception': exc,
                    'transport': self,
                    'protocol': self._protocol,
                })

    def get_write_buffer_limits(self):
        return (self._low_water, self._high_water)

    def _set_write_buffer_limits(self, high=None, low=None):
        if high is None:
            if low is None:
                high = 64 * 1024
            else:
                high = 4 * low
        if low is None:
            low = high // 4

        if not high >= low >= 0:
            raise ValueError(
                f'high ({high!r}) must be >= low ({low!r}) must be >= 0')

        self._high_water = high
        self._low_water = low

    def set_write_buffer_limits(self, high=None, low=None):
        self._set_write_buffer_limits(high=high, low=low)
        self._maybe_pause_protocol()

    def get_write_buffer_size(self):
        raise NotImplementedError