Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _read_packet(self):
"""Read a single packet from the data source.
Returns
-------
dict-like object
"""
assert self._socket is not None
# Reads the header to get the payload length, then reads the payload.
header_buf = util.receive(self._socket, dsi.header_len)
header = dsi.header.parse(header_buf)
payload_buf = util.receive(self._socket, header.payload_length)
return dsi.packet.parse(header_buf + payload_buf)
def _read_packet(self):
"""Read a single packet from the data source.
Returns
-------
dict-like object
"""
assert self._socket is not None
# Reads the header to get the payload length, then reads the payload.
header_buf = util.receive(self._socket, dsi.header_len)
header = dsi.header.parse(header_buf)
payload_buf = util.receive(self._socket, header.payload_length)
return dsi.packet.parse(header_buf + payload_buf)