diff options
| author | Dana Powers <dana.powers@gmail.com> | 2015-03-02 17:49:07 -0800 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2015-03-02 17:49:07 -0800 |
| commit | e13eb0de8903541ec30050a8e7dd8a168752cea3 (patch) | |
| tree | f36a86f711253155439cc71c07bf8f7859822d32 | |
| parent | 7dea5599b64e6e59718b35eb2ce423c12b283799 (diff) | |
| parent | 04979248afa82af77d02f76fbf12456bb83ae334 (diff) | |
| download | kafka-python-e13eb0de8903541ec30050a8e7dd8a168752cea3.tar.gz | |
Merge pull request #324 from chmduquesne/master
Properly destroying the objects that contain daemonized threads
| -rw-r--r-- | kafka/producer/base.py | 6 | ||||
| -rw-r--r-- | kafka/util.py | 3 |
2 files changed, 9 insertions, 0 deletions
diff --git a/kafka/producer/base.py b/kafka/producer/base.py index 5b41bc9..695f195 100644 --- a/kafka/producer/base.py +++ b/kafka/producer/base.py @@ -127,6 +127,7 @@ class Producer(object): self.async = async self.req_acks = req_acks self.ack_timeout = ack_timeout + self.stopped = False if codec is None: codec = CODEC_NONE @@ -212,3 +213,8 @@ class Producer(object): if self.proc.is_alive(): self.proc.terminate() + self.stopped = True + + def __del__(self): + if not self.stopped: + self.stop() diff --git a/kafka/util.py b/kafka/util.py index 14d2b2c..78c3607 100644 --- a/kafka/util.py +++ b/kafka/util.py @@ -151,3 +151,6 @@ class ReentrantTimer(object): # noinspection PyAttributeOutsideInit self.timer = None self.fn = None + + def __del__(self): + self.stop() |