Read a single message from the connection.

        Re-assemble data frames if the message is fragmented.

        Return ``None`` when the closing handshake is started.
893
async def read_message(self) -> Optional[Data]:
        
        frame = await self.read_data_frame(max_size=self.max_size)

        # A close frame was received.
        if frame is None:
            return None

        if frame.opcode == OP_TEXT:
            text = True
        elif frame.opcode == OP_BINARY:
            text = False
        else:  # frame.opcode == OP_CONT
            raise WebSocketProtocolError("Unexpected opcode")

        # Shortcut for the common case - no fragmentation
        if frame.fin:
            return frame.data.decode("utf-8") if text else frame.data

        # 5.4. Fragmentation
        chunks: List[Data] = []
        max_size = self.max_size
        if text:
            decoder_factory = codecs.getincrementaldecoder("utf-8")
            # https://github.com/python/typeshed/pull/2752
            decoder = decoder_factory(errors="strict")  # type: ignore
            if max_size is None:

                def append(frame: Frame) -> None:
                    nonlocal chunks
                    chunks.append(decoder.decode(frame.data, frame.fin))

            else:

                def append(frame: Frame) -> None:
                    nonlocal chunks, max_size
                    chunks.append(decoder.decode(frame.data, frame.fin))
                    max_size -= len(frame.data)

        else:
            if max_size is None:

                def append(frame: Frame) -> None:
                    nonlocal chunks
                    chunks.append(frame.data)

            else:

                def append(frame: Frame) -> None:
                    nonlocal chunks, max_size
                    chunks.append(frame.data)
                    max_size -= len(frame.data)

        append(frame)

        while not frame.fin:
            frame = await self.read_data_frame(max_size=max_size)
            if frame is None:
                raise WebSocketProtocolError("Incomplete fragmented message")
            if frame.opcode != OP_CONT:
                raise WebSocketProtocolError("Unexpected opcode")
            append(frame)

        # mypy cannot figure out that chunks have the proper type.
        return ("" if text else b"").join(chunks)
489288
def on_message(self, new_data):
        """"""
        # https://github.com/socketio/engine.io-protocol

        LOGGER.debug("new frame [%r]", new_data)
        try:
            what = int(new_data[0])
            data = new_data[1:]
            data = data and from_json(data)
            if what == 0:
                self.ping_interval = float(data["pingInterval"]) / 1000
                LOGGER.debug("adjusted ping interval")
                return

            if what == 1:
                LOGGER.debug("received close")
                self.reraise(IOError("Connection closed remotely"))
                return

            if what == 3:
                self.__lastpong = time.time()
                LOGGER.debug("received a pong")
                return

            if what == 4:
                self._on_frame(data)
                return

            if what == 6:
                LOGGER.debug("received noop")
                self.send_message("5")
                return

            LOGGER.debug("unhandled message: [%d] [%r]", what, data)
        except Exception as ex:
            self.reraise(ex)
852013
def _read_message(self):
        
        # first 4 bytes is Big-Endian payload length
        payload_info = self._read_bytes_from_socket(4)
        read_len = unpack(">I", payload_info)[0]

        # now read the payload
        payload = self._read_bytes_from_socket(read_len)

        # pylint: disable=no-member
        message = cast_channel_pb2.CastMessage()
        message.ParseFromString(payload)

        return message
752603
def _read_msg(session):
    
    if len(session.data) == session.message_length:
        # Data Already completely read.  Return
        return True

    try:
        data = session.socket.recv(session.message_length - len(session.data))
        if len(data) == 0:
            raise PushException("No Data on Socket!")
        session.data += data
    except ssl.SSLError:
        # This can happen when select gets triggered
        # for an SSL socket and data has not yet been
        # read.  Wait for it to get triggered again.
        return False

    # Whether or not all data was read.
    return len(session.data) == session.message_length
239240
def read(self, timeout=None):
        '''
        
        '''
        # If currently locked, another greenlet is trying to read, so yield
        # control and then return none. Required if a Connection is configured
        # to be synchronous, a sync callback is trying to read, and there's
        # another read loop running read_frames. Without it, the run loop will
        # release the lock but then immediately acquire it again. Yielding
        # control in the reading thread after bytes are read won't fix
        # anything, because it's quite possible the bytes read resulted in a
        # frame that satisfied the synchronous callback, and so this needs to
        # return immediately to first check the current status of synchronous
        # callbacks before attempting to read again.
        if self._read_lock.locked():
            self._read_wait.wait(timeout)
            return None

        self._read_lock.acquire()
        try:
            return super(GeventTransport, self).read(timeout=timeout)
        finally:
            self._read_lock.release()
            self._read_wait.set()
            self._read_wait.clear()
336008
def _read(self):
    """"""
    with sw("read_response"):
      with catch_websocket_connection_errors():
        response_str = self._sock.recv()
    if not response_str:
      raise ProtocolError("Got an empty response from SC2.")
    response = sc_pb.Response()
    with sw("parse_response"):
      response.ParseFromString(response_str)
    return response
281941
def wait_for_at_least_one_message(self, channel):
        

        unpacker = msgpack.Unpacker(encoding='utf-8')

        while True:
            try:
                start = time.time()
                chunk = self.ssh_channel[channel].recv(1024)
                end = time.time()
                self.read_speeds.append( len(chunk) / (end-start) )
                if len(self.read_speeds) > 20:
                    self.read_speeds = self.read_speeds[10:]

                if chunk == b'':
                    # happens only when connection broke. If nothing is to be received, it hangs instead.
                    self.connection_error(channel, 'Connection broken w')
                    return False
            except Exception as error:
                self.connection_error(channel, error)
                raise

            unpacker.feed(chunk)

            messages = [m for m in unpacker]
            if messages:
                return messages
1270287
def read(message):
    """"""
    require_compatible_version(message.physt_compatible)

    # Currently the only implementation
    a_dict = _dict_from_v0342(message)
    return create_from_dict(a_dict, "Message")
923048
def read(self, timeout=None):
        '''
        
        '''
        # NOTE: copying over this comment from Connection, because there is
        # knowledge captured here, even if the details are stale
        # Because of the timer callback to dataRead when we re-buffered,
        # there's a chance that in between we've lost the socket. If that's
        # the case, just silently return as some code elsewhere would have
        # already notified us. That bug could be fixed by improving the
        # message reading so that we consume all possible messages and ensure
        # that only a partial message was rebuffered, so that we can rely on
        # the next read event to read the subsequent message.
        if not hasattr(self, '_sock'):
            return None

        # This is sort of a hack because we're faking that data is ready, but
        # it works for purposes of supporting timeouts
        if timeout:
            if self._heartbeat_timeout:
                self._heartbeat_timeout.delete()
            self._heartbeat_timeout = \
                event.timeout(timeout, self._sock_read_cb, self._sock)
        elif self._heartbeat_timeout:
            self._heartbeat_timeout.delete()
            self._heartbeat_timeout = None

        return self._sock.read()
2603
def _get_message(self, block=True, timeout=0.1, get_partition_info=None,
                     update_offset=True):
        
        start_at = time.time()
        while self.queue.empty():
            # We're out of messages, go grab some more.
            log.debug('internal queue empty, fetching more messages')
            with FetchContext(self, block, timeout):
                self._fetch()

            if not block or time.time() > (start_at + timeout):
                break

        try:
            partition, message = self.queue.get_nowait()

            if update_offset:
                # Update partition offset
                self.offsets[partition] = message.offset + 1

                # Count, check and commit messages if necessary
                self.count_since_commit += 1
                self._auto_commit()

            if get_partition_info is None:
                get_partition_info = self.partition_info
            if get_partition_info:
                return partition, message
            else:
                return message
        except queue.Empty:
            log.debug('internal queue empty after fetch - returning None')
            return None
651006
def _read_callback(connection_id, data_buffer, data_length_pointer):
    
    wrapped_socket = None
    try:
        wrapped_socket = _connection_refs.get(connection_id)
        if wrapped_socket is None:
            return SecurityConst.errSSLInternal
        base_socket = wrapped_socket.socket

        requested_length = data_length_pointer[0]

        timeout = wrapped_socket.gettimeout()
        error = None
        read_count = 0

        try:
            while read_count < requested_length:
                if timeout is None or timeout >= 0:
                    if not util.wait_for_read(base_socket, timeout):
                        raise socket.error(errno.EAGAIN, 'timed out')

                remaining = requested_length - read_count
                buffer = (ctypes.c_char * remaining).from_address(
                    data_buffer + read_count
                )
                chunk_size = base_socket.recv_into(buffer, remaining)
                read_count += chunk_size
                if not chunk_size:
                    if not read_count:
                        return SecurityConst.errSSLClosedGraceful
                    break
        except (socket.error) as e:
            error = e.errno

            if error is not None and error != errno.EAGAIN:
                data_length_pointer[0] = read_count
                if error == errno.ECONNRESET or error == errno.EPIPE:
                    return SecurityConst.errSSLClosedAbort
                raise

        data_length_pointer[0] = read_count

        if read_count != requested_length:
            return SecurityConst.errSSLWouldBlock

        return 0
    except Exception as e:
        if wrapped_socket is not None:
            wrapped_socket._exception = e
        return SecurityConst.errSSLInternal
749215
def read_handshake(self):
        """"""
        msg = self.read_message()
        pid_dir, _conf, _context = msg["pidDir"], msg["conf"], msg["context"]

        # Write a blank PID file out to the pidDir
        open(join(pid_dir, str(self.pid)), "w").close()
        self.send_message({"pid": self.pid})

        return _conf, _context
862936
def _read_packet(self):
        
        try:
            pos = 0
            while pos < _header.size:
                received = self._transport.recv_into(self._bufview[pos:_header.size-pos])
                if received == 0:
                    raise tds_base.ClosedConnectionError()
                pos += received
        except tds_base.TimeoutError:
            self._session.put_cancel()
            raise
        self._pos = _header.size
        self._type, self._status, self._size, self._session._spid, _ = _header.unpack_from(self._bufview, 0)
        self._have = pos
        while pos < self._size:
            received = self._transport.recv_into(self._bufview[pos:], self._size - pos)
            if received == 0:
                raise tds_base.ClosedConnectionError()
            pos += received
            self._have += received
858150
def receive_message(self):
        
        # split frames
        while len(self._read_buffer) >= INT_SIZE_IN_BYTES:
            frame_length = struct.unpack_from(FMT_LE_INT, self._read_buffer, 0)[0]
            if frame_length > len(self._read_buffer):
                return
            message = ClientMessage(memoryview(self._read_buffer)[:frame_length])
            self._read_buffer = self._read_buffer[frame_length:]
            self._builder.on_message(message)
44257
def _read(self, n):
        
        result = self.sslobj.read(n)

        while len(result) < n:
            s = self.sslobj.read(n - len(result))
            if not s:
                raise IOError('Socket closed')
            result += s

        return result
1023354
def read(self, size):
        
        data = None
        while True:
            try:
                data = self.handle.recv(size)
            except socket.timeout as socket_error:
                self._reconnect(socket_error)
            except socket.error as socket_error:
                # this is fine, just retry
                if socket_error.errno == errno.EINTR:
                    continue
                self._reconnect(IOError)
            if not data:
                self._reconnect(IOError)
            break
        return data
608469
protected DownstreamMessage receiveMessage ()
        throws IOException
    {
        // read in the next message frame (readFrame() can return false meaning it only read part
        // of the frame from the network, in which case we simply call it again because we can't do
        // anything until it has a whole frame; it will throw an exception if it hits EOF or if
        // something goes awry)
        while (!readFrame()) {
            // noop!
        }

        if (_oin == null) {
            // Our object input stream was taken away from us before we could do anything with the
            //  frame, this is equivalent to being interrupted mid-frame, so indicate that.
            throw new InterruptedIOException();
        }

        try {
            int size = _fin.available();
            DownstreamMessage msg = (DownstreamMessage)_oin.readObject();
            if (debugLogMessages()) {
                log.info("RECEIVE " + msg);
            }
            _client.getMessageTracker().messageReceived(false, size, msg, 0);
            return msg;

        } catch (ClassNotFoundException cnfe) {
            throw (IOException) new IOException(
                "Unable to decode incoming message.").initCause(cnfe);
        }
    }