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