Non-blocking network receive.

        Return list of (response, future) tuples

def recv(self):
        """Non-blocking network receive.

        Return list of (response, future) tuples
        """
        responses = self._recv()
        if not responses and self.requests_timed_out():
            log.warning('%s timed out after %s ms. Closing connection.',
                        self, self.config['request_timeout_ms'])
            self.close(error=Errors.RequestTimedOutError(
                'Request timed out after %s ms' %
                self.config['request_timeout_ms']))
            return ()

        # augment respones w/ correlation_id, future, and timestamp
        for i, (correlation_id, response) in enumerate(responses):
            try:
                with self._lock:
                    (future, timestamp) = self.in_flight_requests.pop(correlation_id)
            except KeyError:
                self.close(Errors.KafkaConnectionError('Received unrecognized correlation id'))
                return ()
            latency_ms = (time.time() - timestamp) * 1000
            if self._sensors:
                self._sensors.request_time.record(latency_ms)

            log.debug('%s Response %d (%s ms): %s', self, correlation_id, latency_ms, response)
            responses[i] = (response, future)

        return responses