'Try to read and write to sockets.
This method will also attempt to complete node connections, refresh
stale metadata, and run previously-scheduled tasks.
Arguments:
timeout_ms (int, optional): maximum amount of time to wait (in ms)
for at least one response. Must be non-negative. The actual
timeout will be the minimum of timeout, request timeout and
metadata timeout. Default: request_timeout_ms
future (Future, optional): if provided, blocks until future.is_done
sleep (bool): if True and there is nothing to do (no connections
or requests in flight), will sleep for duration timeout before
returning empty results. Default: False.
Returns:
list: responses received (can be empty)'

def poll(self, timeout_ms=None, future=None, sleep=True, delayed_tasks=True):

    if (timeout_ms is None):
        timeout_ms = self.config['request_timeout_ms']
    responses = []
    while True:
        for node_id in list(self._connecting):
            self._maybe_connect(node_id)
        metadata_timeout_ms = self._maybe_refresh_metadata()
        if delayed_tasks:
            for (task, task_future) in self._delayed_tasks.pop_ready():
                try:
                    result = task()
                except Exception as e:
                    log.error('Task    %s    failed:    %s', task, e)
                    task_future.failure(e)
                else:
                    task_future.success(result)
        if (future and future.is_done):
            timeout = 0
        else:
            idle_connection_timeout_ms = self._idle_expiry_manager.next_check_ms()
            timeout = min(timeout_ms, metadata_timeout_ms, (self._delayed_tasks.next_at() * 1000), idle_connection_timeout_ms, self.config['request_timeout_ms'])
            timeout = max(0, (timeout / 1000.0))
        responses.extend(self._poll(timeout, sleep=sleep))
        if ((not future) or future.is_done):
            break
    return responses
