docs: def receive_bytes(self, data):
          """Process bytes received from the network.
  
          Arguments:
              data (bytes): any length bytes received from a network connection
                  to a kafka broker.
  
          Returns:
              responses (list of (correlation_id, response)): any/all completed
                  responses, decoded from bytes to python objects.
  
          Raises:
               KafkaProtocolError: if the bytes received could not be decoded.
               CorrelationIdError: if the response does not match the request
                   correlation id.
          """
          i = 0
          n = len(data)
          responses = []
          while i < n:
  
              # Not receiving is the state of reading the payload header
              if not self._receiving:
                  bytes_to_read = min(4 - self._header.tell(), n - i)
                  self._header.write(data[i:i+bytes_to_read])
                  i += bytes_to_read
  
                  if self._header.tell() == 4:
                      self._header.seek(0)
                      nbytes = Int32.decode(self._header)
                      # reset buffer and switch state to receiving payload bytes
                      self._rbuffer = KafkaBytes(nbytes)
                      self._receiving = True
                  elif self._header.tell() > 4:
                      raise Errors.KafkaError('this should not happen - are you threading?')
  
              if self._receiving:
                  total_bytes = len(self._rbuffer)
                  staged_bytes = self._rbuffer.tell()
                  bytes_to_read = min(total_bytes - staged_bytes, n - i)
                  self._rbuffer.write(data[i:i+bytes_to_read])
                  i += bytes_to_read
  
                  staged_bytes = self._rbuffer.tell()
                  if staged_bytes > total_bytes:
                      raise Errors.KafkaError('Receive buffer has more bytes than expected?')
  
                  if staged_bytes != total_bytes:
                      break
  
                  self._receiving = False
                  self._rbuffer.seek(0)
                  resp = self._process_response(self._rbuffer)
                  responses.append(resp)
                  self._reset_buffer()
          return responses