this tracks the state of app protocol https git iofj59p init cm conmade dr er eof cl conlost cm connectionmade dr datareceived er eofreceived cl connectionlost client side may pass ssltrue to use a default context in that case the sslcontext passed is none the default is secure for client connections python 3 4 use uptodate strong settings get optional transport information return self sslprotocol getextrainfoname default def setprotocolself protocol self sslprotocol setappprotocolprotocol def getprotocolself return self sslprotocol appprotocol def isclosingself return self closed def closeself if not self closed self closed true self sslprotocol startshutdown else self sslprotocol none def delself warningswarnings if not self closed self closed true warnings warn unclosed transport asyncio sslprotocoltransport object resourcewarning def isreadingself return not self sslprotocol appreadingpaused def pausereadingself self sslprotocol pausereading def resumereadingself self sslprotocol resumereading def setwritebufferlimitsself highnone lownone self sslprotocol setwritebufferlimitshigh low self sslprotocol controlappwriting def getwritebufferlimitsself return self sslprotocol outgoinglowwater self sslprotocol outgoinghighwater def getwritebuffersizeself set the high and lowwater limits for read flow control these two values control when to call the upstream transport s pausereading and resumereading methods if specified the lowwater limit must be less than or equal to the highwater limit neither value can be negative the defaults are implementationspecific if only the highwater limit is given the lowwater limit defaults to an implementationspecific value less than or equal to the highwater limit setting high to zero forces low to zero as well and causes pausereading to be called whenever the buffer becomes nonempty setting low to zero causes resumereading to be called only once the buffer is empty use of zero for either limit is generally suboptimal as it reduces opportunities for doing io and computation concurrently return the current size of the read buffer return self sslprotocol getreadbuffersize property def protocolpausedself required for sendfile fallback pausewritingresumewriting logic return self sslprotocol appwritingpaused def writeself data if not isinstancedata bytes bytearray memoryview raise typeerrorfdata expecting a byteslike instance fgot typedata name if not data return self sslprotocol writeappdatadata def writelinesself listofdata self sslprotocol writeappdatalistofdata 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 for test only sslspecific extra info more info are set when the handshake completes app data write buffering transport ex selectorsockettransport ssl and state machine flow control make fast hasattr check first called when the lowlevel connection is made start the ssl handshake called when the lowlevel connection is lost or closed the argument is an exception object or none the latter meaning a regular eof is received or the connection was aborted or closed just mark the app transport as closed so that its dealloc doesn t complain called when the other end of the lowlevel stream is halfclosed if this returns a false value including none the transport will close itself if it returns a true value closing the transport is up to the protocol handshake flow start handshake timeout count down add extra info that becomes available after handshake shutdown flow outgoing flow incoming flow closenotify closenotify flow control for writes from app socket flow control for reads to app socket flow control for reads from ssl socket flow control for writes to ssl socket called when the lowlevel transport s buffer goes over the highwater mark called when the lowlevel transport s buffer drains below the lowwater mark pragma no cover this tracks the state of app protocol https git io fj59p init cm con_made dr er eof cl con_lost cm connection_made dr data_received er eof_received cl connection_lost client side may pass ssl true to use a default context in that case the sslcontext passed is none the default is secure for client connections python 3 4 use up to date strong settings get optional transport information 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 called with none as its argument 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 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 buffers set the high and low water limits for read flow control these two values control when to call the upstream transport s pause_reading and resume_reading 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_reading to be called whenever the buffer becomes non empty setting low to zero causes resume_reading 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 read buffer required for sendfile fallback pause_writing resume_writing logic 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 raises exc notimplementederror right now 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 for test only buffer size passed to read ssl specific extra info more info are set when the handshake completes app data write buffering transport ex selectorsockettransport ssl and state machine set when connection_lost called flow control make fast hasattr check first called when the low level connection is made start the ssl handshake called when the low level connection is lost or closed the argument is an exception object or none the latter meaning a regular eof is received or the connection was aborted or closed just mark the app transport as closed so that its __dealloc__ doesn t complain called when the other end of the low level stream is half closed if this returns a false value including none the transport will close itself if it returns a true value closing the transport is up to the protocol handshake flow start handshake timeout count down add extra info that becomes available after handshake shutdown flow outgoing flow incoming flow close_notify close_notify flow control for writes from app socket flow control for reads to app socket flow control for reads from ssl socket flow control for writes to ssl socket called when the low level transport s buffer goes over the high water mark called when the low level transport s buffer drains below the low water mark
import collections
import enum
import warnings
try:
    import ssl
except ImportError:  
    ssl = None

from . import constants
from . import exceptions
from . import protocols
from . import transports
from .log import logger

if ssl is not None:
    SSLAgainErrors = (ssl.SSLWantReadError, ssl.SSLSyscallError)


class SSLProtocolState(enum.Enum):
    UNWRAPPED = "UNWRAPPED"
    DO_HANDSHAKE = "DO_HANDSHAKE"
    WRAPPED = "WRAPPED"
    FLUSHING = "FLUSHING"
    SHUTDOWN = "SHUTDOWN"


class AppProtocolState(enum.Enum):
    
    STATE_INIT = "STATE_INIT"
    STATE_CON_MADE = "STATE_CON_MADE"
    STATE_EOF = "STATE_EOF"
    STATE_CON_LOST = "STATE_CON_LOST"


def _create_transport_context(server_side, server_hostname):
    if server_side:
        raise ValueError('Server side SSL needs a valid SSLContext')

    sslcontext = ssl.create_default_context()
    if not server_hostname:
        sslcontext.check_hostname = False
    return sslcontext


def add_flowcontrol_defaults(high, low, kb):
    if high is None:
        if low is None:
            hi = kb * 1024
        else:
            lo = low
            hi = 4 * lo
    else:
        hi = high
    if low is None:
        lo = hi // 4
    else:
        lo = low

    if not hi >= lo >= 0:
        raise ValueError('high (%r) must be >= low (%r) must be >= 0' %
                         (hi, lo))

    return hi, lo


class _SSLProtocolTransport(transports._FlowControlMixin,
                            transports.Transport):

    _start_tls_compatible = True
    _sendfile_compatible = constants._SendfileMode.FALLBACK

    def __init__(self, loop, ssl_protocol):
        self._loop = loop
        self._ssl_protocol = ssl_protocol
        self._closed = False

    def get_extra_info(self, name, default=None):
        
        return self._ssl_protocol._get_extra_info(name, default)

    def set_protocol(self, protocol):
        self._ssl_protocol._set_app_protocol(protocol)

    def get_protocol(self):
        return self._ssl_protocol._app_protocol

    def is_closing(self):
        return self._closed

    def close(self):
        
        if not self._closed:
            self._closed = True
            self._ssl_protocol._start_shutdown()
        else:
            self._ssl_protocol = None

    def __del__(self, _warnings=warnings):
        if not self._closed:
            self._closed = True
            _warnings.warn(
                "unclosed transport <asyncio._SSLProtocolTransport "
                "object>", ResourceWarning)

    def is_reading(self):
        return not self._ssl_protocol._app_reading_paused

    def pause_reading(self):
        
        self._ssl_protocol._pause_reading()

    def resume_reading(self):
        
        self._ssl_protocol._resume_reading()

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

    def get_write_buffer_limits(self):
        return (self._ssl_protocol._outgoing_low_water,
                self._ssl_protocol._outgoing_high_water)

    def get_write_buffer_size(self):
        
        return self._ssl_protocol._get_write_buffer_size()

    def set_read_buffer_limits(self, high=None, low=None):
        
        self._ssl_protocol._set_read_buffer_limits(high, low)
        self._ssl_protocol._control_ssl_reading()

    def get_read_buffer_limits(self):
        return (self._ssl_protocol._incoming_low_water,
                self._ssl_protocol._incoming_high_water)

    def get_read_buffer_size(self):
        
        return self._ssl_protocol._get_read_buffer_size()

    @property
    def _protocol_paused(self):
        
        return self._ssl_protocol._app_writing_paused

    def write(self, data):
        
        if not isinstance(data, (bytes, bytearray, memoryview)):
            raise TypeError(f"data: expecting a bytes-like instance, "
                            f"got {type(data).__name__}")
        if not data:
            return
        self._ssl_protocol._write_appdata((data,))

    def writelines(self, list_of_data):
        
        self._ssl_protocol._write_appdata(list_of_data)

    def write_eof(self):
        
        raise NotImplementedError

    def can_write_eof(self):
        
        return False

    def abort(self):
        
        self._closed = True
        if self._ssl_protocol is not None:
            self._ssl_protocol._abort()

    def _force_close(self, exc):
        self._closed = True
        self._ssl_protocol._abort(exc)

    def _test__append_write_backlog(self, data):
        
        self._ssl_protocol._write_backlog.append(data)
        self._ssl_protocol._write_buffer_size += len(data)


class SSLProtocol(protocols.BufferedProtocol):
    max_size = 256 * 1024   

    _handshake_start_time = None
    _handshake_timeout_handle = None
    _shutdown_timeout_handle = None

    def __init__(self, loop, app_protocol, sslcontext, waiter,
                 server_side=False, server_hostname=None,
                 call_connection_made=True,
                 ssl_handshake_timeout=None,
                 ssl_shutdown_timeout=None):
        if ssl is None:
            raise RuntimeError("stdlib ssl module not available")

        self._ssl_buffer = bytearray(self.max_size)
        self._ssl_buffer_view = memoryview(self._ssl_buffer)

        if ssl_handshake_timeout is None:
            ssl_handshake_timeout = constants.SSL_HANDSHAKE_TIMEOUT
        elif ssl_handshake_timeout <= 0:
            raise ValueError(
                f"ssl_handshake_timeout should be a positive number, "
                f"got {ssl_handshake_timeout}")
        if ssl_shutdown_timeout is None:
            ssl_shutdown_timeout = constants.SSL_SHUTDOWN_TIMEOUT
        elif ssl_shutdown_timeout <= 0:
            raise ValueError(
                f"ssl_shutdown_timeout should be a positive number, "
                f"got {ssl_shutdown_timeout}")

        if not sslcontext:
            sslcontext = _create_transport_context(
                server_side, server_hostname)

        self._server_side = server_side
        if server_hostname and not server_side:
            self._server_hostname = server_hostname
        else:
            self._server_hostname = None
        self._sslcontext = sslcontext
        
        
        self._extra = dict(sslcontext=sslcontext)

        
        self._write_backlog = collections.deque()
        self._write_buffer_size = 0

        self._waiter = waiter
        self._loop = loop
        self._set_app_protocol(app_protocol)
        self._app_transport = None
        self._app_transport_created = False
        
        self._transport = None
        self._ssl_handshake_timeout = ssl_handshake_timeout
        self._ssl_shutdown_timeout = ssl_shutdown_timeout
        
        self._incoming = ssl.MemoryBIO()
        self._outgoing = ssl.MemoryBIO()
        self._state = SSLProtocolState.UNWRAPPED
        self._conn_lost = 0  
        if call_connection_made:
            self._app_state = AppProtocolState.STATE_INIT
        else:
            self._app_state = AppProtocolState.STATE_CON_MADE
        self._sslobj = self._sslcontext.wrap_bio(
            self._incoming, self._outgoing,
            server_side=self._server_side,
            server_hostname=self._server_hostname)

        

        self._ssl_writing_paused = False

        self._app_reading_paused = False

        self._ssl_reading_paused = False
        self._incoming_high_water = 0
        self._incoming_low_water = 0
        self._set_read_buffer_limits()
        self._eof_received = False

        self._app_writing_paused = False
        self._outgoing_high_water = 0
        self._outgoing_low_water = 0
        self._set_write_buffer_limits()
        self._get_app_transport()

    def _set_app_protocol(self, app_protocol):
        self._app_protocol = app_protocol
        
        if (hasattr(app_protocol, 'get_buffer') and
                isinstance(app_protocol, protocols.BufferedProtocol)):
            self._app_protocol_get_buffer = app_protocol.get_buffer
            self._app_protocol_buffer_updated = app_protocol.buffer_updated
            self._app_protocol_is_buffer = True
        else:
            self._app_protocol_is_buffer = False

    def _wakeup_waiter(self, exc=None):
        if self._waiter is None:
            return
        if not self._waiter.cancelled():
            if exc is not None:
                self._waiter.set_exception(exc)
            else:
                self._waiter.set_result(None)
        self._waiter = None

    def _get_app_transport(self):
        if self._app_transport is None:
            if self._app_transport_created:
                raise RuntimeError('Creating _SSLProtocolTransport twice')
            self._app_transport = _SSLProtocolTransport(self._loop, self)
            self._app_transport_created = True
        return self._app_transport

    def connection_made(self, transport):
        
        self._transport = transport
        self._start_handshake()

    def connection_lost(self, exc):
        
        self._write_backlog.clear()
        self._outgoing.read()
        self._conn_lost += 1

        
        
        if self._app_transport is not None:
            self._app_transport._closed = True

        if self._state != SSLProtocolState.DO_HANDSHAKE:
            if (
                self._app_state == AppProtocolState.STATE_CON_MADE or
                self._app_state == AppProtocolState.STATE_EOF
            ):
                self._app_state = AppProtocolState.STATE_CON_LOST
                self._loop.call_soon(self._app_protocol.connection_lost, exc)
        self._set_state(SSLProtocolState.UNWRAPPED)
        self._transport = None
        self._app_transport = None
        self._app_protocol = None
        self._wakeup_waiter(exc)

        if self._shutdown_timeout_handle:
            self._shutdown_timeout_handle.cancel()
            self._shutdown_timeout_handle = None
        if self._handshake_timeout_handle:
            self._handshake_timeout_handle.cancel()
            self._handshake_timeout_handle = None

    def get_buffer(self, n):
        want = n
        if want <= 0 or want > self.max_size:
            want = self.max_size
        if len(self._ssl_buffer) < want:
            self._ssl_buffer = bytearray(want)
            self._ssl_buffer_view = memoryview(self._ssl_buffer)
        return self._ssl_buffer_view

    def buffer_updated(self, nbytes):
        self._incoming.write(self._ssl_buffer_view[:nbytes])

        if self._state == SSLProtocolState.DO_HANDSHAKE:
            self._do_handshake()

        elif self._state == SSLProtocolState.WRAPPED:
            self._do_read()

        elif self._state == SSLProtocolState.FLUSHING:
            self._do_flush()

        elif self._state == SSLProtocolState.SHUTDOWN:
            self._do_shutdown()

    def eof_received(self):
        
        self._eof_received = True
        try:
            if self._loop.get_debug():
                logger.debug("%r received EOF", self)

            if self._state == SSLProtocolState.DO_HANDSHAKE:
                self._on_handshake_complete(ConnectionResetError)

            elif self._state == SSLProtocolState.WRAPPED:
                self._set_state(SSLProtocolState.FLUSHING)
                if self._app_reading_paused:
                    return True
                else:
                    self._do_flush()

            elif self._state == SSLProtocolState.FLUSHING:
                self._do_write()
                self._set_state(SSLProtocolState.SHUTDOWN)
                self._do_shutdown()

            elif self._state == SSLProtocolState.SHUTDOWN:
                self._do_shutdown()

        except Exception:
            self._transport.close()
            raise

    def _get_extra_info(self, name, default=None):
        if name in self._extra:
            return self._extra[name]
        elif self._transport is not None:
            return self._transport.get_extra_info(name, default)
        else:
            return default

    def _set_state(self, new_state):
        allowed = False

        if new_state == SSLProtocolState.UNWRAPPED:
            allowed = True

        elif (
            self._state == SSLProtocolState.UNWRAPPED and
            new_state == SSLProtocolState.DO_HANDSHAKE
        ):
            allowed = True

        elif (
            self._state == SSLProtocolState.DO_HANDSHAKE and
            new_state == SSLProtocolState.WRAPPED
        ):
            allowed = True

        elif (
            self._state == SSLProtocolState.WRAPPED and
            new_state == SSLProtocolState.FLUSHING
        ):
            allowed = True

        elif (
            self._state == SSLProtocolState.FLUSHING and
            new_state == SSLProtocolState.SHUTDOWN
        ):
            allowed = True

        if allowed:
            self._state = new_state

        else:
            raise RuntimeError(
                'cannot switch state from {} to {}'.format(
                    self._state, new_state))

    

    def _start_handshake(self):
        if self._loop.get_debug():
            logger.debug("%r starts SSL handshake", self)
            self._handshake_start_time = self._loop.time()
        else:
            self._handshake_start_time = None

        self._set_state(SSLProtocolState.DO_HANDSHAKE)

        
        self._handshake_timeout_handle = \
            self._loop.call_later(self._ssl_handshake_timeout,
                                  self._check_handshake_timeout)

        self._do_handshake()

    def _check_handshake_timeout(self):
        if self._state == SSLProtocolState.DO_HANDSHAKE:
            msg = (
                f"SSL handshake is taking longer than "
                f"{self._ssl_handshake_timeout} seconds: "
                f"aborting the connection"
            )
            self._fatal_error(ConnectionAbortedError(msg))

    def _do_handshake(self):
        try:
            self._sslobj.do_handshake()
        except SSLAgainErrors:
            self._process_outgoing()
        except ssl.SSLError as exc:
            self._on_handshake_complete(exc)
        else:
            self._on_handshake_complete(None)

    def _on_handshake_complete(self, handshake_exc):
        if self._handshake_timeout_handle is not None:
            self._handshake_timeout_handle.cancel()
            self._handshake_timeout_handle = None

        sslobj = self._sslobj
        try:
            if handshake_exc is None:
                self._set_state(SSLProtocolState.WRAPPED)
            else:
                raise handshake_exc

            peercert = sslobj.getpeercert()
        except Exception as exc:
            self._set_state(SSLProtocolState.UNWRAPPED)
            if isinstance(exc, ssl.CertificateError):
                msg = 'SSL handshake failed on verifying the certificate'
            else:
                msg = 'SSL handshake failed'
            self._fatal_error(exc, msg)
            self._wakeup_waiter(exc)
            return

        if self._loop.get_debug():
            dt = self._loop.time() - self._handshake_start_time
            logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)

        
        self._extra.update(peercert=peercert,
                           cipher=sslobj.cipher(),
                           compression=sslobj.compression(),
                           ssl_object=sslobj)
        if self._app_state == AppProtocolState.STATE_INIT:
            self._app_state = AppProtocolState.STATE_CON_MADE
            self._app_protocol.connection_made(self._get_app_transport())
        self._wakeup_waiter()
        self._do_read()

    

    def _start_shutdown(self):
        if (
            self._state in (
                SSLProtocolState.FLUSHING,
                SSLProtocolState.SHUTDOWN,
                SSLProtocolState.UNWRAPPED
            )
        ):
            return
        if self._app_transport is not None:
            self._app_transport._closed = True
        if self._state == SSLProtocolState.DO_HANDSHAKE:
            self._abort()
        else:
            self._set_state(SSLProtocolState.FLUSHING)
            self._shutdown_timeout_handle = self._loop.call_later(
                self._ssl_shutdown_timeout,
                self._check_shutdown_timeout
            )
            self._do_flush()

    def _check_shutdown_timeout(self):
        if (
            self._state in (
                SSLProtocolState.FLUSHING,
                SSLProtocolState.SHUTDOWN
            )
        ):
            self._transport._force_close(
                exceptions.TimeoutError('SSL shutdown timed out'))

    def _do_flush(self):
        self._do_read()
        self._set_state(SSLProtocolState.SHUTDOWN)
        self._do_shutdown()

    def _do_shutdown(self):
        try:
            if not self._eof_received:
                self._sslobj.unwrap()
        except SSLAgainErrors:
            self._process_outgoing()
        except ssl.SSLError as exc:
            self._on_shutdown_complete(exc)
        else:
            self._process_outgoing()
            self._call_eof_received()
            self._on_shutdown_complete(None)

    def _on_shutdown_complete(self, shutdown_exc):
        if self._shutdown_timeout_handle is not None:
            self._shutdown_timeout_handle.cancel()
            self._shutdown_timeout_handle = None

        if shutdown_exc:
            self._fatal_error(shutdown_exc)
        else:
            self._loop.call_soon(self._transport.close)

    def _abort(self):
        self._set_state(SSLProtocolState.UNWRAPPED)
        if self._transport is not None:
            self._transport.abort()

    

    def _write_appdata(self, list_of_data):
        if (
            self._state in (
                SSLProtocolState.FLUSHING,
                SSLProtocolState.SHUTDOWN,
                SSLProtocolState.UNWRAPPED
            )
        ):
            if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
                logger.warning('SSL connection is closed')
            self._conn_lost += 1
            return

        for data in list_of_data:
            self._write_backlog.append(data)
            self._write_buffer_size += len(data)

        try:
            if self._state == SSLProtocolState.WRAPPED:
                self._do_write()

        except Exception as ex:
            self._fatal_error(ex, 'Fatal error on SSL protocol')

    def _do_write(self):
        try:
            while self._write_backlog:
                data = self._write_backlog[0]
                count = self._sslobj.write(data)
                data_len = len(data)
                if count < data_len:
                    self._write_backlog[0] = data[count:]
                    self._write_buffer_size -= count
                else:
                    del self._write_backlog[0]
                    self._write_buffer_size -= data_len
        except SSLAgainErrors:
            pass
        self._process_outgoing()

    def _process_outgoing(self):
        if not self._ssl_writing_paused:
            data = self._outgoing.read()
            if len(data):
                self._transport.write(data)
        self._control_app_writing()

    

    def _do_read(self):
        if (
            self._state not in (
                SSLProtocolState.WRAPPED,
                SSLProtocolState.FLUSHING,
            )
        ):
            return
        try:
            if not self._app_reading_paused:
                if self._app_protocol_is_buffer:
                    self._do_read__buffered()
                else:
                    self._do_read__copied()
                if self._write_backlog:
                    self._do_write()
                else:
                    self._process_outgoing()
            self._control_ssl_reading()
        except Exception as ex:
            self._fatal_error(ex, 'Fatal error on SSL protocol')

    def _do_read__buffered(self):
        offset = 0
        count = 1

        buf = self._app_protocol_get_buffer(self._get_read_buffer_size())
        wants = len(buf)

        try:
            count = self._sslobj.read(wants, buf)

            if count > 0:
                offset = count
                while offset < wants:
                    count = self._sslobj.read(wants - offset, buf[offset:])
                    if count > 0:
                        offset += count
                    else:
                        break
                else:
                    self._loop.call_soon(self._do_read)
        except SSLAgainErrors:
            pass
        if offset > 0:
            self._app_protocol_buffer_updated(offset)
        if not count:
            
            self._call_eof_received()
            self._start_shutdown()

    def _do_read__copied(self):
        chunk = b'1'
        zero = True
        one = False

        try:
            while True:
                chunk = self._sslobj.read(self.max_size)
                if not chunk:
                    break
                if zero:
                    zero = False
                    one = True
                    first = chunk
                elif one:
                    one = False
                    data = [first, chunk]
                else:
                    data.append(chunk)
        except SSLAgainErrors:
            pass
        if one:
            self._app_protocol.data_received(first)
        elif not zero:
            self._app_protocol.data_received(b''.join(data))
        if not chunk:
            
            self._call_eof_received()
            self._start_shutdown()

    def _call_eof_received(self):
        try:
            if self._app_state == AppProtocolState.STATE_CON_MADE:
                self._app_state = AppProtocolState.STATE_EOF
                keep_open = self._app_protocol.eof_received()
                if keep_open:
                    logger.warning('returning true from eof_received() '
                                   'has no effect when using ssl')
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as ex:
            self._fatal_error(ex, 'Error calling eof_received()')

    

    def _control_app_writing(self):
        size = self._get_write_buffer_size()
        if size >= self._outgoing_high_water and not self._app_writing_paused:
            self._app_writing_paused = True
            try:
                self._app_protocol.pause_writing()
            except (KeyboardInterrupt, SystemExit):
                raise
            except BaseException as exc:
                self._loop.call_exception_handler({
                    'message': 'protocol.pause_writing() failed',
                    'exception': exc,
                    'transport': self._app_transport,
                    'protocol': self,
                })
        elif size <= self._outgoing_low_water and self._app_writing_paused:
            self._app_writing_paused = False
            try:
                self._app_protocol.resume_writing()
            except (KeyboardInterrupt, SystemExit):
                raise
            except BaseException as exc:
                self._loop.call_exception_handler({
                    'message': 'protocol.resume_writing() failed',
                    'exception': exc,
                    'transport': self._app_transport,
                    'protocol': self,
                })

    def _get_write_buffer_size(self):
        return self._outgoing.pending + self._write_buffer_size

    def _set_write_buffer_limits(self, high=None, low=None):
        high, low = add_flowcontrol_defaults(
            high, low, constants.FLOW_CONTROL_HIGH_WATER_SSL_WRITE)
        self._outgoing_high_water = high
        self._outgoing_low_water = low

    

    def _pause_reading(self):
        self._app_reading_paused = True

    def _resume_reading(self):
        if self._app_reading_paused:
            self._app_reading_paused = False

            def resume():
                if self._state == SSLProtocolState.WRAPPED:
                    self._do_read()
                elif self._state == SSLProtocolState.FLUSHING:
                    self._do_flush()
                elif self._state == SSLProtocolState.SHUTDOWN:
                    self._do_shutdown()
            self._loop.call_soon(resume)

    

    def _control_ssl_reading(self):
        size = self._get_read_buffer_size()
        if size >= self._incoming_high_water and not self._ssl_reading_paused:
            self._ssl_reading_paused = True
            self._transport.pause_reading()
        elif size <= self._incoming_low_water and self._ssl_reading_paused:
            self._ssl_reading_paused = False
            self._transport.resume_reading()

    def _set_read_buffer_limits(self, high=None, low=None):
        high, low = add_flowcontrol_defaults(
            high, low, constants.FLOW_CONTROL_HIGH_WATER_SSL_READ)
        self._incoming_high_water = high
        self._incoming_low_water = low

    def _get_read_buffer_size(self):
        return self._incoming.pending

    

    def pause_writing(self):
        
        assert not self._ssl_writing_paused
        self._ssl_writing_paused = True

    def resume_writing(self):
        
        assert self._ssl_writing_paused
        self._ssl_writing_paused = False
        self._process_outgoing()

    def _fatal_error(self, exc, message='Fatal error on transport'):
        if self._transport:
            self._transport._force_close(exc)

        if isinstance(exc, OSError):
            if self._loop.get_debug():
                logger.debug("%r: %s", self, message, exc_info=True)
        elif not isinstance(exc, exceptions.CancelledError):
            self._loop.call_exception_handler({
                'message': message,
                'exception': exc,
                'transport': self._transport,
                'protocol': self,
            })