From cfae9e3fa3432fad6bbd97c1d81f8ea4bc0ee363 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 30 Dec 2015 11:50:42 -0800 Subject: Remove unnecessary calls in KafkaClient._poll - Dont process connections; outer poll() loop does this now - Only recv connections that select says are readable --- kafka/client_async.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) (limited to 'kafka/client_async.py') diff --git a/kafka/client_async.py b/kafka/client_async.py index ca81214..eaa5ef0 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -264,22 +264,22 @@ class KafkaClient(object): def _poll(self, timeout): # select on reads across all connected sockets, blocking up to timeout - sockets = [conn._sock for conn in six.itervalues(self._conns) - if (conn.state is ConnectionStates.CONNECTED and - conn.in_flight_requests)] - if sockets: - select.select(sockets, [], [], timeout) + sockets = dict([(conn._sock, conn) + for conn in six.itervalues(self._conns) + if (conn.state is ConnectionStates.CONNECTED + and conn.in_flight_requests)]) + if not sockets: + return [] + + ready, _, _ = select.select(list(sockets.keys()), [], [], timeout) responses = [] # list, not iterator, because inline callbacks may add to self._conns - for conn in list(self._conns.values()): - if conn.state is ConnectionStates.CONNECTING: - conn.connect() - - if conn.in_flight_requests: - response = conn.recv() # This will run callbacks / errbacks - if response: - responses.append(response) + for sock in ready: + conn = sockets[sock] + response = conn.recv() # Note: conn.recv runs callbacks / errbacks + if response: + responses.append(response) return responses def in_flight_request_count(self, node_id=None): -- cgit v1.2.1