docs: def _recv(self):
          """Take all available bytes from socket, return list of any responses from parser"""
          recvd = []
          self._lock.acquire()
          if not self._can_send_recv():
              log.warning('%s cannot recv: socket not connected', self)
              self._lock.release()
              return ()
  
          while len(recvd) < self.config['sock_chunk_buffer_count']:
              try:
                  data = self._sock.recv(self.config['sock_chunk_bytes'])
                  # We expect socket.recv to raise an exception if there are no
                  # bytes available to read from the socket in non-blocking mode.
                  # but if the socket is disconnected, we will get empty data
                  # without an exception raised
                  if not data:
                      log.error('%s: socket disconnected', self)
                      self._lock.release()
                      self.close(error=Errors.KafkaConnectionError('socket disconnected'))
                      return []
                  else:
                      recvd.append(data)
  
              except SSLWantReadError:
                  break
              except ConnectionError as e:
                  if six.PY2 and e.errno == errno.EWOULDBLOCK:
                      break
                  log.exception('%s: Error receiving network data'
                                ' closing socket', self)
                  self._lock.release()
                  self.close(error=Errors.KafkaConnectionError(e))
                  return []
              except BlockingIOError:
                  if six.PY3:
                      break
                  self._lock.release()
                  raise
  
          recvd_data = b''.join(recvd)
          if self._sensors:
              self._sensors.bytes_received.record(len(recvd_data))
  
          try:
              responses = self._protocol.receive_bytes(recvd_data)
          except Errors.KafkaProtocolError as e:
              self._lock.release()
              self.close(e)
              return []
          else:
              self._lock.release()
              return responses