summaryrefslogtreecommitdiff
path: root/kafka/client_async.py
diff options
context:
space:
mode:
authorDana Powers <dana.powers@gmail.com>2016-04-04 18:55:17 -0700
committerDana Powers <dana.powers@gmail.com>2016-04-07 08:28:00 -0700
commit931373478e30a9d44b89fad6491136222441e929 (patch)
treefc5379c6bfb2d49a38d3bd681dd1634007ace363 /kafka/client_async.py
parent7f042327c8ee8089c5d57c91ccdf2ac7e9ebf1b8 (diff)
downloadkafka-python-931373478e30a9d44b89fad6491136222441e929.tar.gz
Use conn.connecting() checks to support multiple connecting states
Diffstat (limited to 'kafka/client_async.py')
-rw-r--r--kafka/client_async.py6
1 files changed, 3 insertions, 3 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 907ee0c..e51e3d4 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -118,7 +118,7 @@ class KafkaClient(object):
log.debug("Attempting to bootstrap via node at %s:%s", host, port)
bootstrap = BrokerConnection(host, port, afi, **self.config)
bootstrap.connect()
- while bootstrap.state is ConnectionStates.CONNECTING:
+ while bootstrap.connecting():
bootstrap.connect()
if bootstrap.state is not ConnectionStates.CONNECTED:
bootstrap.close()
@@ -164,7 +164,7 @@ class KafkaClient(object):
self._conns[node_id] = BrokerConnection(host, broker.port, afi,
**self.config)
state = self._conns[node_id].connect()
- if state is ConnectionStates.CONNECTING:
+ if self._conns[node_id].connecting():
self._connecting.add(node_id)
# Whether CONNECTED or DISCONNECTED, we need to remove from connecting
@@ -251,7 +251,7 @@ class KafkaClient(object):
time_waited_ms = time.time() - (conn.last_attempt or 0)
if conn.state is ConnectionStates.DISCONNECTED:
return max(self.config['reconnect_backoff_ms'] - time_waited_ms, 0)
- elif conn.state is ConnectionStates.CONNECTING:
+ elif conn.connecting():
return 0
else:
return 999999999