openstack/deb-python-kafka
Code Issues Proposed changes

Add size and correlation id decoding to try/except block in BrokerConnection

Browse Source
This commit is contained in:
Dana Powers
2015-12-01 16:21:08 -08:00
committed by Zack Dever
parent 254c17e39f
commit 7a6c51bf2e
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
kafka/conn.py
View File

@@ -88,12 +88,13 @@ class BrokerConnection(local):
# instead we read directly from the socket fd buffer # instead we read directly from the socket fd buffer
# alternatively, we could read size bytes into a separate buffer # alternatively, we could read size bytes into a separate buffer
# and decode from that buffer (and verify buffer is empty afterwards) # and decode from that buffer (and verify buffer is empty afterwards)
size = Int32.decode(self._read_fd)
recv_correlation_id = Int32.decode(self._read_fd)
assert correlation_id == recv_correlation_id
try: try:
size = Int32.decode(self._read_fd)
recv_correlation_id = Int32.decode(self._read_fd)
if correlation_id != recv_correlation_id:
raise RuntimeError('Correlation ids do not match!')
response = response_type.decode(self._read_fd) response = response_type.decode(self._read_fd)
except socket.error as e: except (RuntimeError, socket.error) as e:
log.exception("Error in BrokerConnection.recv()") log.exception("Error in BrokerConnection.recv()")
self.close() self.close()
return None return None