Send or receive a chunk of data.

        :arg size:  Amount of data.  0 indicates EOT.
963565
def streamWrite(self, size):
        
        size = int(size)
        if size == 0:
            self._close()
            return

        self._sendResult(dict(message="writing...", stream=True, size=size))
        data = sys.stdin.read(size)
        self.stream.write(data)
927928
async def _receive_chunk(self, chunk):
        
        self.__log_debug('< %s', chunk)

        # common
        if isinstance(chunk, DataChunk):
            await self._receive_data_chunk(chunk)
        elif isinstance(chunk, SackChunk):
            await self._receive_sack_chunk(chunk)
        elif isinstance(chunk, ForwardTsnChunk):
            await self._receive_forward_tsn_chunk(chunk)
        elif isinstance(chunk, HeartbeatChunk):
            ack = HeartbeatAckChunk()
            ack.params = chunk.params
            await self._send_chunk(ack)
        elif isinstance(chunk, AbortChunk):
            self.__log_debug('x Association was aborted by remote party')
            self._set_state(self.State.CLOSED)
        elif isinstance(chunk, ShutdownChunk):
            self._t2_cancel()
            self._set_state(self.State.SHUTDOWN_RECEIVED)
            ack = ShutdownAckChunk()
            await self._send_chunk(ack)
            self._t2_start(ack)
            self._set_state(self.State.SHUTDOWN_ACK_SENT)
        elif (isinstance(chunk, ShutdownCompleteChunk) and
              self._association_state == self.State.SHUTDOWN_ACK_SENT):
            self._t2_cancel()
            self._set_state(self.State.CLOSED)
        elif (isinstance(chunk, ReconfigChunk) and
              self._association_state == self.State.ESTABLISHED):
            for param in chunk.params:
                cls = RECONFIG_PARAM_TYPES.get(param[0])
                if cls:
                    await self._receive_reconfig_param(cls.parse(param[1]))

        # server
        elif isinstance(chunk, InitChunk) and self.is_server:
            self._last_received_tsn = tsn_minus_one(chunk.initial_tsn)
            self._reconfig_response_seq = tsn_minus_one(chunk.initial_tsn)
            self._remote_verification_tag = chunk.initiate_tag
            self._ssthresh = chunk.advertised_rwnd
            self._get_extensions(chunk.params)

            self.__log_debug('- Peer supports %d outbound streams, %d max inbound streams',
                             chunk.outbound_streams, chunk.inbound_streams)
            self._inbound_streams_count = min(chunk.outbound_streams, self._inbound_streams_max)
            self._outbound_streams_count = min(self._outbound_streams_count, chunk.inbound_streams)

            ack = InitAckChunk()
            ack.initiate_tag = self._local_verification_tag
            ack.advertised_rwnd = self._advertised_rwnd
            ack.outbound_streams = self._outbound_streams_count
            ack.inbound_streams = self._inbound_streams_max
            ack.initial_tsn = self._local_tsn
            self._set_extensions(ack.params)

            # generate state cookie
            cookie = pack('!L', self._get_timestamp())
            cookie += hmac.new(self._hmac_key, cookie, 'sha1').digest()
            ack.params.append((SCTP_STATE_COOKIE, cookie))
            await self._send_chunk(ack)
        elif isinstance(chunk, CookieEchoChunk) and self.is_server:
            # check state cookie MAC
            cookie = chunk.body
            if (len(cookie) != COOKIE_LENGTH or
               hmac.new(self._hmac_key, cookie[0:4], 'sha1').digest() != cookie[4:]):
                self.__log_debug('x State cookie is invalid')
                return

            # check state cookie lifetime
            now = self._get_timestamp()
            stamp = unpack_from('!L', cookie)[0]
            if stamp < now - COOKIE_LIFETIME or stamp > now:
                self.__log_debug('x State cookie has expired')
                error = ErrorChunk()
                error.params.append((SCTP_CAUSE_STALE_COOKIE, b'\x00' * 8))
                await self._send_chunk(error)
                return

            ack = CookieAckChunk()
            await self._send_chunk(ack)
            self._set_state(self.State.ESTABLISHED)

        # client
        elif isinstance(chunk, InitAckChunk) and self._association_state == self.State.COOKIE_WAIT:
            # cancel T1 timer and process chunk
            self._t1_cancel()
            self._last_received_tsn = tsn_minus_one(chunk.initial_tsn)
            self._reconfig_response_seq = tsn_minus_one(chunk.initial_tsn)
            self._remote_verification_tag = chunk.initiate_tag
            self._ssthresh = chunk.advertised_rwnd
            self._get_extensions(chunk.params)

            self.__log_debug('- Peer supports %d outbound streams, %d max inbound streams',
                             chunk.outbound_streams, chunk.inbound_streams)
            self._inbound_streams_count = min(chunk.outbound_streams, self._inbound_streams_max)
            self._outbound_streams_count = min(self._outbound_streams_count, chunk.inbound_streams)

            echo = CookieEchoChunk()
            for k, v in chunk.params:
                if k == SCTP_STATE_COOKIE:
                    echo.body = v
                    break
            await self._send_chunk(echo)

            # start T1 timer and enter COOKIE-ECHOED state
            self._t1_start(echo)
            self._set_state(self.State.COOKIE_ECHOED)
        elif (isinstance(chunk, CookieAckChunk) and
              self._association_state == self.State.COOKIE_ECHOED):
            # cancel T1 timer and enter ESTABLISHED state
            self._t1_cancel()
            self._set_state(self.State.ESTABLISHED)
        elif (isinstance(chunk, ErrorChunk) and
              self._association_state in [self.State.COOKIE_WAIT, self.State.COOKIE_ECHOED]):
            self._t1_cancel()
            self._set_state(self.State.CLOSED)
            self.__log_debug('x Could not establish association')
            return
1300789
def send(self, data):
        """"""
        log.debug('Sending %s' % data)
        if not self._socket:
            log.warn('No connection')
            return
        self._socket.send_bytes(data.encode('utf-8'))
1296474
def read(self, size=None):
        
        data = self.rfile.read(size)
        self.bytes_read += len(data)
        self._check_length()
        return data
309247
def _send_size(self):
        "  "
        rows, cols = _get_size(sys.stdout.fileno())
        self._send_packet({
            'cmd': 'size',
            'data': [rows, cols]
        })
1296204
def read(self, size=None):
        
        data = EMPTY

        if size == 0:
            return data

        while True:
            if size and len(data) >= size:
                return data

            if not self.buffer:
                self._fetch()
                if not self.buffer:
                    # EOF
                    return data

            if size:
                remaining = size - len(data)
                data += self.buffer[:remaining]
                self.buffer = self.buffer[remaining:]
            else:
                data += self.buffer
                self.buffer = EMPTY
153941
def recvall(sock, size):
    
    data = b""
    while len(data) < size:
        s = sock.recv(size - len(data))
        data += s
        if not s:
            raise RuntimeError("did not receive data from socket "
                               "(size {}, got only {!r})".format(size, data))
    return data
608244
def _read_socket(self, size):
        
        value = b''
        while len(value) < size:
            data = self.connection.recv(size - len(value))
            if not data:
                break
            value += data

        # If we got less data than we requested, the server disconnected.
        if len(value) < size:
            raise socket.error()

        return value
809054
def receive(self, data):
        
        self._incoming_segments_queue.put(data)
        if not self._in_handshake():
            self._flush_incoming_buffer()
651947
def stream_download(self, chunk_size: Optional[int] = None, callback: Optional[Callable] = None) -> AsyncIterator[bytes]:
        
        pass
12097
def write(self, data):
        """"""
        # If nobody is connected, act like all data was written anyway.
        if self.connected is None:
            return 0
        data = to_bytes_safe(data)
        size = len(data)
        remaining = size
        while remaining:
            count = self._abstract_socket.write(data)
            remaining -= count
            if remaining:
                data = data[count:]
        return size
1178914
def receive(self, data):
        
        self.log_debug("Received: %s" % (data))
        message =  self.make_message(data)
        self.trigger("receive", data=data, message=message)
        return message
711870
def read_and_write(self, reader, writer, chunk_size):
        
        data = reader.recv(chunk_size)
        if len(data) == 0:
            return True
        writer.sendall(data)
1374295
def send_chunk(type, content = nil)
      chunk = Chunk.new(type, content).to_s
      socket.write chunk
    end
507761
def send_by_packet(self, data):
        
        total_sent = 0
        while total_sent < PACKET_SIZE:
            sent = self.sock.send(data[total_sent:])
            if sent == 0:
                raise RuntimeError("socket connection broken")
            total_sent += sent
        return total_sent
495882
def drain(self):
        
        data = self._stream.getvalue()
        if len(data):
            yield from self._protocol.send(data)
        self._stream = io.BytesIO(b'')
783420
def send(self, data: bytes = b""):
        
        self.input.extend(data)
        self._process()